### Start Postgres Docker

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/dbos/_templates/dbos-db-starter/README.md

Starts a Postgres database using Docker.

```shell
export PGPASSWORD=dbos
python3 start_postgres_docker.py
```

--------------------------------

### Minimal DBOS Configuration and Launch

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/README.md

Configure DBOS with a database URL and launch the application. This is the most basic setup required to start a DBOS application.

```python
from dbos import DBOS, DBOSConfig

config = DBOSConfig(
    database_url="postgresql://user:pass@localhost/dbos"
)
dbos = DBOS(config=config)
dbos.launch()
```

--------------------------------

### Install DBOS Cloud CLI

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/dbos/_templates/dbos-db-starter/README.md

Installs the DBOS Cloud CLI using npm.

```shell
npm i -g @dbos-inc/dbos-cloud
```

--------------------------------

### DBOS Configuration File Example

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

An example of a dbos-config.yaml file, showing database URL, application version, executor ID, runtime configuration, and tracing settings.

```yaml
# dbos-config.yaml
database_url: postgresql://user:pass@localhost/dbos
application_version: "1.0.0"
executor_id: "worker-1"

runtimeConfig:
  max_executor_threads: 50

tracing:
  otel_exporter_otlp_endpoint: "http://localhost:4318"
  otel_exporter_otlp_protocol: "http/protobuf"
```

--------------------------------

### Start a local Postgres database for testing

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/DEVELOPING.md

Starts a local PostgreSQL database using Docker for testing purposes.

```bash
export PGPASSWORD=dbos
python3 dbos/_templates/dbos-db-starter/start_postgres_docker.py
```

--------------------------------

### DBOS.launch()

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Launches the DBOS runtime. Must be called after DBOS construction but before starting workflows. Initializes the database, starts background threads, and registers decorators.

```APIDOC
## DBOS.launch()

### Description
Launches the DBOS runtime. Must be called after DBOS construction but before starting workflows. Initializes the database, starts background threads, and registers decorators.

### Method
`classmethod`

### Endpoint
N/A (Class Method)

### Parameters
None

### Request Example
```python
DBOS.launch()
```

### Response
#### Success Response (200)
None (This method does not return a value).

#### Response Example
None
```

--------------------------------

### Configure OpenTelemetry Tracing

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Set up OpenTelemetry tracing for monitoring. Requires the `otel` extra to be installed (`pip install dbos[otel]`).

```python
config = DBOSConfig(
    database_url="...",
    tracing={
        "otel_exporter_otlp_endpoint": "http://localhost:4318",
        "otel_exporter_otlp_protocol": "http/protobuf",
    },
)
```

--------------------------------

### DBOS CLI Commands

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/dbos/_templates/dbos-db-starter/README.md

Commands to migrate and start the DBOS application locally.

```shell
dbos migrate
dbos start
```

--------------------------------

### Install pdm

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/DEVELOPING.md

Installs the pdm package manager using a curl command.

```bash
curl -sSL https://pdm-project.org/install-pdm.py | python3 -
```

--------------------------------

### Queue Constructor Example

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/queue.md

Creates a queue for managing workflow execution with specified concurrency.

```python
queue = Queue("payment_queue", concurrency=10)

@DBOS.step()
def process_payment(order_id: str):
    charge_card(order_id)

@DBOS.workflow()
def batch_payments(order_ids: list[str]):
    handles = []
    for oid in order_ids:
        handle = queue.enqueue(process_payment, oid)
        handles.append(handle)
    return [h.get_result() for h in handles]
```

--------------------------------

### Configure OpenTelemetry Tracing

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Set up distributed tracing by providing OpenTelemetry exporter endpoint, protocol, and service name in the DBOSConfig. Ensure you install the 'otel' extra: `pip install dbos[otel]`.

```python
config = DBOSConfig(
    database_url="...",
    tracing={
        "otel_exporter_otlp_endpoint": "http://localhost:4317",
        "otel_exporter_otlp_protocol": "grpc",  # or "http/protobuf"
        "service_name": "my-dbos-app",
    },
)
```

--------------------------------

### Example StepOptions Usage

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/types.md

Demonstrates how to use StepOptions with the @DBOS.step decorator to configure retries and timeouts for a payment charging function.

```python
@DBOS.step(name="charge_card", max_retries=3, timeout_seconds=30)
def charge_payment(amount: float):
    return payment_api.charge(amount)
```

--------------------------------

### Install Python dependencies with pdm

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/DEVELOPING.md

Installs project dependencies and sets up pre-commit hooks using pdm.

```bash
pdm install
pdm run pre-commit install
```

--------------------------------

### Launch DBOS Runtime

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Call the launch method to initialize the DBOS runtime after constructing the DBOS instance. This must be done before starting any workflows.

```python
DBOS.launch()
```

--------------------------------

### Example Kafka Consumer Workflow

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/decorators.md

Illustrates setting up a Kafka consumer workflow to process messages from 'orders' and 'refunds' topics. It includes Kafka configuration and message processing logic based on the topic.

```python
import json
from dbos import DBOS, KafkaMessage

kafka_config = {
    "bootstrap.servers": "localhost:9092",
    "security.protocol": "PLAINTEXT",
}

@DBOS.kafka_consumer(kafka_config, ["orders", "refunds"])
@DBOS.workflow()
def process_event(msg: KafkaMessage):
    event = json.loads(msg.value.decode())
    
    if msg.topic == "orders":
        process_order(event)
    elif msg.topic == "refunds":
        process_refund(event)
```

--------------------------------

### Start a Kafka Consumer Workflow

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/decorators.md

Use the @DBOS.kafka_consumer decorator to start a workflow that consumes Kafka messages with exactly-once semantics. It requires Kafka configuration, a list of topics, and an optional consumer group ID.

```python
@DBOS.kafka_consumer(
    config: dict[str, Any],
    topics: list[str],
    consumer_group_id: Optional[str] = None,
)
@DBOS.workflow()
def my_consumer(msg: KafkaMessage):
    pass
```

--------------------------------

### Start a Synchronous Workflow

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Use `start_workflow` to initiate a workflow synchronously. It returns a handle to poll for results. Ensure the workflow function is registered.

```python
handle = DBOS.start_workflow(order_workflow, "order_123")
result = handle.get_result()  # Blocks until complete
```

--------------------------------

### Debouncer Workflow Example

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/types.md

Illustrates how to use the Debouncer class within a DBOS workflow to enqueue requests and retrieve their results.

```python
debouncer = Debouncer("index_updates", delay_seconds=5, max_batch_size=100)

@DBOS.workflow()
def handle_update(event: dict):
    request_id = event["id"]
    debouncer.enqueue(request_id, event)
    result = debouncer.get_result(request_id)
    return result
```

--------------------------------

### DBOS CLI Commands

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/utilities.md

Basic commands for the DBOS CLI, including running migrations, starting the application, and executing Python scripts.

```bash
# Run migrations
dbos migrate

# Start the app
dbos start

# Run a Python script
dbos run script.py

```

--------------------------------

### Example Scheduled Workflows

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/decorators.md

Demonstrates scheduling a daily cleanup workflow and a frequent sync workflow using the @DBOS.scheduled decorator with different cron expressions and timezones.

```python
from datetime import datetime

@DBOS.scheduled("0 0 * * *", cron_timezone="America/New_York")
@DBOS.workflow()
def daily_cleanup(scheduled_time: datetime, actual_time: datetime):
    logger.info(f"Running cleanup at {actual_time}")
    delete_old_records()

@DBOS.scheduled("*/5 * * * *")  # Every 5 minutes
@DBOS.workflow()
def frequent_sync(scheduled_time: datetime, actual_time: datetime):
    sync_external_data()
```

--------------------------------

### Handling DBOSInitializationError

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/errors.md

Illustrates catching a DBOSInitializationError during DBOS Transact setup, typically due to invalid configuration or connection issues.

```python
from dbos import DBOS, DBOSConfig, DBOSInitializationError

try:
    config = DBOSConfig(database_url="invalid://url")
    dbos = DBOS(config=config)
    dbos.launch()
except DBOSInitializationError as e:
    print(f"Failed to initialize: {e.message}")
    exit(1)
```

--------------------------------

### start_workflow()

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Starts a workflow synchronously and returns immediately with a handle to poll for results. This method is suitable for blocking operations where immediate workflow initiation is required.

```APIDOC
## start_workflow()

### Description
Starts a workflow synchronously and returns immediately with a handle to poll for results.

### Method Signature
```python
@classmethod
def start_workflow(
    cls,
    workflow_fn: Callable[P, R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> WorkflowHandle[R]
```

### Parameters
* **workflow_fn** (`Callable[P, R]`) - Required - The workflow function to execute.
* **args** (`P.args`) - Required - Positional arguments to the workflow.
* **kwargs** (`P.kwargs`) - Required - Keyword arguments to the workflow.

### Returns
* `WorkflowHandle[R]` – A handle to retrieve the workflow result and status.

### Throws
* `DBOSConflictingWorkflowError` - A workflow with the same ID was already started.
* `DBOSException` - Workflow function not found or not registered.

### Example
```python
handle = DBOS.start_workflow(order_workflow, "order_123")
result = handle.get_result()  # Blocks until complete
```
```

--------------------------------

### Get Configuration Value with Default

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Retrieve a configuration value using its key, providing a default value if the key is not found.

```python
version = config.get("application_version", "unknown")
```

--------------------------------

### start_workflow_async()

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Asynchronously starts a workflow and returns a handle for polling results. This non-blocking method is ideal for asynchronous applications.

```APIDOC
## start_workflow_async()

### Description
Asynchronously starts a workflow and returns a handle for polling results.

### Method Signature
```python
@classmethod
async def start_workflow_async(
    cls,
    workflow_fn: Callable[P, Coroutine[Any, Any, R]],
    *args: P.args,
    **kwargs: P.kwargs,
) -> WorkflowHandleAsync[R]
```

### Parameters
* **workflow_fn** (`Callable[P, Coroutine[Any, Any, R]]`) - Required - Async workflow function.
* **args** (`P.args`) - Required - Positional arguments.
* **kwargs** (`P.kwargs`) - Required - Keyword arguments.

### Returns
* `WorkflowHandleAsync[R]` – Async handle for the workflow.

### Example
```python
handle = await DBOS.start_workflow_async(async_workflow, "param1")
result = await handle.get_result()
```
```

--------------------------------

### Start an Asynchronous Workflow

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Use `start_workflow_async` for asynchronous workflow initiation. It returns an async handle for polling results. The workflow function must be an async coroutine.

```python
handle = await DBOS.start_workflow_async(async_workflow, "param1")
result = await handle.get_result()
```

--------------------------------

### Enqueue Task with MaxPriority

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/types.md

Example of enqueuing a task with the highest possible priority using SetEnqueueOptions.

```python
with SetEnqueueOptions({"priority": MaxPriority}):
    queue.enqueue(high_priority_task)
```

--------------------------------

### @DBOS.kafka_consumer()

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/decorators.md

Starts a Kafka consumer workflow with exactly-once semantics. Each message triggers a new workflow execution.

```APIDOC
## @DBOS.kafka_consumer()

### Description
Starts a Kafka consumer workflow with exactly-once semantics. Each message triggers a new workflow execution.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Method
Decorator

### Endpoint
N/A

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
#### config
- **config** (`dict[str, Any]`) - Required - Kafka client config (bootstrap.servers, etc.)
- **topics** (`list[str]`) - Required - List of topic names to consume
- **consumer_group_id** (`Optional[str]`) - Optional - Consumer group ID; defaults to function name. Defaults to None.

### Returns
Decorated function.

### Message Parameter
The workflow function receives a `KafkaMessage` with:
- `topic`: Topic name
- `partition`: Partition number
- `offset`: Message offset
- `key`: Message key (bytes)
- `value`: Message payload (bytes)
- `headers`: Message headers dict

### Behavior
- One workflow execution per message
- Exactly-once semantics using workflow IDs
- Automatically handles offset tracking

### Example
```python
import json
from dbos import DBOS, KafkaMessage

kafka_config = {
    "bootstrap.servers": "localhost:9092",
    "security.protocol": "PLAINTEXT",
}

@DBOS.kafka_consumer(kafka_config, ["orders", "refunds"])
@DBOS.workflow()
def process_event(msg: KafkaMessage):
    event = json.loads(msg.value.decode())
    
    if msg.topic == "orders":
        process_order(event)
    elif msg.topic == "refunds":
        process_refund(event)
```
```

--------------------------------

### Reliable Billing Workflow Example

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/README.md

This example demonstrates how to build a reliable billing workflow that durably waits for a notification from a payments service, processing it exactly-once.

```python
@DBOS.workflow()
def billing_workflow():
  ... # Calculate the charge, then submit the bill to a payments service
  payment_status = DBOS.recv(PAYMENT_STATUS, timeout=payment_service_timeout)
  if payment_status is not None and payment_status == "paid":
      ... # Handle a successful payment.
  else:
      ... # Handle a failed payment or timeout.

```

--------------------------------

### Check DBOS Version using Package

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/utilities.md

Access the DBOS version directly from the installed package.

```python
import dbos
print(dbos.__version__)
```

--------------------------------

### DBOS Queues Example

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/README.md

Demonstrates how to enqueue tasks into a DBOS queue for background processing within a durable workflow.

```python
from dbos import DBOS, Queue

queue = Queue("example_queue")

@DBOS.step()
def process_task(task):
  ...

@DBOS.workflow()
def process_tasks(tasks):
  task_handles = []
  # Enqueue each task so all tasks are processed concurrently.
  for task in tasks:
    handle = queue.enqueue(process_task, task)
    task_handles.append(handle)
  # Wait for each task to complete and retrieve its result.
  # Return the results of all tasks.
  return [handle.get_result() for handle in task_handles]
```

--------------------------------

### Durable Scheduling Example (Cron)

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/README.md

Demonstrates scheduling a workflow to run at a specific interval using cron syntax.

```python
@DBOS.scheduled('* * * * *') # crontab syntax to run once every minute
@DBOS.workflow()
def example_scheduled_workflow(scheduled_time: datetime, actual_time: datetime):
    DBOS.logger.info("I am a workflow scheduled to run once a minute.")
```

--------------------------------

### Programmatic Workflow Management Example

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/README.md

Shows how to use the DBOSClient to query, pause, resume, or restart workflow executions programmatically.

```python
# Create a DBOS client connected to your Postgres database.
client = DBOSClient(database_url)
# Find all workflows that errored between 3:00 and 5:00 AM UTC on 2025-04-22.
workflows = client.list_workflows(status="ERROR", 
  start_time="2025-04-22T03:00:00Z", end_time="2025-04-22T05:00:00Z")
for workflow in workflows:
    # Check which workflows failed due to an outage in a service called from Step 2.
    steps = client.list_workflow_steps(workflow)
    if len(steps) >= 3 and isinstance(steps[2]["error"], ServiceOutage):
        # To recover from the outage, restart those workflows from Step 2.
        DBOS.fork_workflow(workflow.workflow_id, 2)
```

--------------------------------

### Get Workflow ID (Asynchronous)

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Retrieves the unique identifier for an asynchronously started workflow. Use this when you need to reference a workflow by its ID.

```python
def get_workflow_id(self) -> str:
    return self.workflow_id
```

--------------------------------

### Get Workflow Status (Synchronous)

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Retrieves the current status of a synchronously started workflow without blocking. Useful for monitoring progress.

```python
status = handle.get_status()
print(f"Status: {status.status}, Steps: {len(status.steps)}")
```

--------------------------------

### Deploy to DBOS Cloud

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/dbos/_templates/dbos-db-starter/README.md

Deploys the application to DBOS Cloud.

```shell
dbos-cloud app deploy
```

--------------------------------

### Get Workflow ID (Synchronous)

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Retrieves the unique identifier for a synchronously started workflow. Use this when you need to reference a workflow by its ID.

```python
workflow_id = handle.get_workflow_id()
```

--------------------------------

### Exactly-Once Event Processing (Webhooks)

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/README.md

Example of handling incoming webhooks with exactly-once processing using an idempotency key.

```python
def handle_message(request: Request) -> None:
  event_id = request.body["event_id"]
  # Use the event ID as an idempotency key to start the workflow exactly-once
  with SetWorkflowID(event_id):
    # Start the workflow in the background, then acknowledge the event
    DBOS.start_workflow(message_workflow, request.body["event"])
```

--------------------------------

### Instantiate and Launch DBOS

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Create a DBOS instance using configuration loaded from environment variables and then launch the DBOS runtime.

```python
from dbos import DBOS, DBOSConfig

config = DBOSConfig.from_env()
dbos = DBOS(config=config)
dbos.launch()
```

--------------------------------

### Get Workflow Result (Asynchronous)

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Asynchronously waits for an asynchronously started workflow to complete and returns its result. Specify a polling interval to control how often the status is checked.

```python
result = await handle.get_result()
```

--------------------------------

### Get Workflow Result (Synchronous)

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Blocks until a synchronously started workflow completes and returns its result. Specify a polling interval to control how often the status is checked. Throws DBOSException if the workflow fails.

```python
result = handle.get_result()
print(f"Order total: {result}")
```

--------------------------------

### Initialize DBOSConfig with Keyword Arguments

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Directly create a DBOSConfig object by passing configuration parameters as keyword arguments. This method allows for explicit setting of various options like database URLs, executor IDs, and application versions.

```python
config = DBOSConfig(
    database_url="postgresql://user:pass@localhost:5432/dbos",
    executor_id="executor-1",
    application_version="1.2.3",
)
dbos = DBOS(config=config)
```

--------------------------------

### Example: Pydantic Validation for Payment Requests

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/decorators.md

This example demonstrates using @pydantic_args_validator with a PaymentRequest model, including an EmailStr type for email validation. Invalid arguments will be rejected before the step executes.

```python
from pydantic import BaseModel, EmailStr

class PaymentRequest(BaseModel):
    order_id: str
    amount: float
    email: EmailStr

@pydantic_args_validator(PaymentRequest)
@DBOS.step(max_retries=3)
def charge_card(request: PaymentRequest):
    payment_api.charge(request.order_id, request.amount)
    send_receipt(request.email)
```

--------------------------------

### Load DBOS Configuration from Environment and YAML

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Load DBOS configuration by combining settings from environment variables and a YAML configuration file.

```python
config = DBOSConfig.from_env()  # Loads from YAML + env vars
```

--------------------------------

### Initialize DBOSClient

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/client.md

Create an instance of DBOSClient to access DBOS workflow data. This client is for read/write operations and does not run migrations or launch DBOS. Configure it with URLs for system and application databases.

```python
client = DBOSClient(
    system_database_url="postgresql://user:pass@localhost/dbos",
    application_database_url="postgresql://user:pass@localhost/app",
)
```

--------------------------------

### Create DBOSConfig using a Configuration Builder

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Instantiate a ConfigBuilder to fluently set database URLs and other configuration options before building the final DBOSConfig object. This is useful for programmatic configuration.

```python
config = DBOSConfig.new().set_database_url("postgresql://...").build()
```

--------------------------------

### backfill_schedule()

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/client.md

Backfills a schedule with missed executions between the specified start and end times.

```APIDOC
## backfill_schedule()

### Description
Backfills a schedule with missed executions between the specified times.

### Parameters
#### Path Parameters
- **schedule_name** (`str`) - Required - Name of the schedule
- **start_time** (`datetime`) - Required - Start of backfill window
- **end_time** (`datetime`) - Required - End of backfill window

### Example
```python
from datetime import datetime, timedelta

client.backfill_schedule(
    "hourly_sync",
    start_time=datetime(2025, 1, 1),
    end_time=datetime(2025, 1, 2),
)
```
```

--------------------------------

### Run DBOS Database Migrations via CLI

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/utilities.md

This command-line interface command initiates the database migration process.

```bash
dbos migrate
```

--------------------------------

### backfill_schedule_async()

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/client.md

Asynchronously backfills a schedule with missed executions between the specified start and end times.

```APIDOC
## backfill_schedule_async()

### Description
Asynchronously backfills a schedule.

### Parameters
#### Path Parameters
- **schedule_name** (`str`) - Required - Name of the schedule
- **start_time** (`datetime`) - Required - Start of backfill window
- **end_time** (`datetime`) - Required - End of backfill window
```

--------------------------------

### Development Runtime Configuration from Environment Variables

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Load DBOS configuration from environment variables prefixed with DBOS_* for development.

```python
config = DBOSConfig.from_env()  # Loads from DBOS_* env vars
```

--------------------------------

### Handling DBOSConflictingWorkflowError

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/errors.md

Demonstrates how to catch and handle a DBOSConflictingWorkflowError when starting a workflow with a pre-existing ID.

```python
from dbos import DBOS, SetWorkflowID, DBOSConflictingWorkflowError

try:
    with SetWorkflowID("fixed_id"):
        handle = DBOS.start_workflow(my_workflow)
except DBOSConflictingWorkflowError as e:
    print(f"Workflow already started: {e.message}")
```

--------------------------------

### Create DBOSConfig from Environment Variables

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Use this method to create a DBOSConfig object by reading settings from environment variables prefixed with DBOS_*. Ensure the necessary environment variables like DBOS_DATABASE_URL are set before calling.

```python
import os

os.environ["DBOS_DATABASE_URL"] = "postgresql://user:pass@localhost/myapp"
config = DBOSConfig.from_env()
```

--------------------------------

### kafka_consumer()

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Decorator that starts a Kafka consumer workflow that runs once per message with exactly-once semantics.

```APIDOC
## kafka_consumer()

### Description
Decorator that starts a Kafka consumer workflow that runs once per message with exactly-once semantics.

### Method Signature
```python
@classmethod
def kafka_consumer(
    cls,
    config: dict[str, Any],
    topics: list[str],
    consumer_group_id: Optional[str] = None,
) -> Callable[[F], F]
```

### Parameters
#### Arguments
- **config** (`dict[str, Any]`): Kafka configuration dict (e.g., bootstrap.servers).
- **topics** (`list[str]`): List of topic names to consume from.
- **consumer_group_id** (`Optional[str]`, optional): Consumer group ID; defaults to function name.

### Returns
A decorator function.

### Example
```python
kafka_config = {"bootstrap.servers": "localhost:9092"}

@DBOS.kafka_consumer(kafka_config, ["orders"])
@DBOS.workflow()
def process_order(msg: KafkaMessage):
    order = json.loads(msg.value.decode())
    # Process order durably with exactly-once semantics
```
```

--------------------------------

### Initialize DBOSConfig with Default Serializer

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Instantiate DBOSConfig with a database URL. The DefaultSerializer (JSON) is used automatically.

```python
config = DBOSConfig(database_url="...")
# Uses DefaultSerializer automatically
```

--------------------------------

### Full DBOS Configuration Options

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/README.md

Configure DBOS with all available options, including system database URL, application version, executor ID, serializer, tracing endpoints, and runtime configurations like max threads.

```python
config = DBOSConfig(
    database_url="postgresql://user:pass@localhost/app",
    system_database_url="postgresql://user:pass@localhost/dbos_system",
    application_version="1.2.3",
    executor_id="worker-1",
    serializer=DBOSPortableJSONSerializer(),
    tracing={
        "otel_exporter_otlp_endpoint": "http://localhost:4318",
    },
    runtimeConfig={
        "max_executor_threads": 50,
    },
)
```

--------------------------------

### Production Runtime Configuration

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Configure DBOS for a production environment, loading settings from environment variables and system information.

```python
config = DBOSConfig(
    database_url=os.environ["DATABASE_URL"],
    application_version=os.environ.get("APP_VERSION", "prod"),
    executor_id=socket.gethostname(),
    tracing={
        "otel_exporter_otlp_endpoint": "https://otel.example.com",
        "otel_exporter_otlp_protocol": "http/protobuf",
    },
    runtimeConfig={
        "max_executor_threads": 100,
    },
)
```

--------------------------------

### Get Workflow Status

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Retrieve the status of a workflow using its ID with `get_workflow_status`. Returns `None` if the workflow is not found.

```python
status = DBOS.get_workflow_status(workflow_id)
if status and status.status == "SUCCEEDED":
    print(f"Workflow completed: {status.output}")
```

--------------------------------

### DBOS CLI Environment Variables

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/utilities.md

Demonstrates setting DBOS environment variables to configure database URL and application version before running a command.

```bash
export DBOS_DATABASE_URL="postgresql://..."
export DBOS_APPVERSION="1.2.3"
dbos migrate

```

--------------------------------

### Define DBOS Workflow with Type Hinting

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/utilities.md

Use Python's typing module for DBOS function signatures, including lists, dictionaries, and optional parameters.

```python
from typing import Optional, List, Dict, Any
from dbos import DBOS

@DBOS.workflow()
def process_orders(
    orders: List[Dict[str, Any]],
    customer_id: str,
    options: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    return {"processed": len(orders)}
```

--------------------------------

### Create SQLAlchemyDatasource Instance

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/datasource.md

Instantiate SQLAlchemyDatasource with a name and database connection details. Use either a database URL or an existing SQLAlchemy engine.

```python
ds = SQLAlchemyDatasource(
    name="app_db",
    url="postgresql://user:pass@localhost/myapp",
    engine_kwargs={"pool_size": 10, "max_overflow": 5},
)
```

--------------------------------

### Create AsyncSQLAlchemyDatasource with URL

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/datasource.md

Instantiate an AsyncSQLAlchemyDatasource using a database connection URL. This is useful for setting up a new asynchronous database connection.

```python
async_ds = AsyncSQLAlchemyDatasource(
    name="async_db",
    url="postgresql+asyncpg://user:pass@localhost/myapp",
)
```

--------------------------------

### Testing Configuration with In-Memory SQLite

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Configure DBOS for testing using an in-memory SQLite database and specific identifiers.

```python
import tempfile

config = DBOSConfig(
    database_url="sqlite:///:memory:",
    executor_id="test-runner",
    application_version="test",
)
```

--------------------------------

### Get DBOS Application Version

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Fetch the application version, which can be set by DBOS Cloud or manually, using the DBOS__APPVERSION environment variable.

```python
import os

version = os.environ.get("DBOS__APPVERSION", "")
```

--------------------------------

### Configure Conductor URL and Key

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Specify the DBOS Conductor service URL and the API key for authentication.

```python
config = DBOSConfig(
    database_url="...",
    conductor_url="https://conductor.example.com",
    conductor_key="api_key_xyz",
)
```

--------------------------------

### Asynchronously Get Workflow Status

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/client.md

Asynchronously retrieve the status of a specific workflow by its ID. This method is suitable for use within async functions.

```python
async def get_workflow_async(self, workflow_id: str) -> Optional[WorkflowStatus]:
    # ... implementation details ...
    pass
```

--------------------------------

### Priority and Partitioning Configuration

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/queue.md

Configure whether priority-based execution and queue partitioning are enabled.

```APIDOC
## Queue Priority and Partitioning

### Description
Configure and check the status of priority-based execution and queue partitioning.

### Properties

#### `priority_enabled`
- **Description**: Gets whether priority-based execution is enabled on this queue.
- **Returns**: `bool` – True if priority is enabled, False otherwise.

#### `partition_queue`
- **Description**: Gets whether queue partitioning is enabled.
- **Returns**: `bool` – True if partitioning is enabled, False otherwise.

### Methods

#### `get_priority_enabled_async()`
- **Description**: Asynchronously checks if priority-based execution is enabled on this queue.
- **Returns**: `bool` – True if priority is enabled, False otherwise.

#### `get_partition_queue_async()`
- **Description**: Asynchronously checks if queue partitioning is enabled.
- **Returns**: `bool` – True if partitioning is enabled, False otherwise.
```

--------------------------------

### Get Specific Workflow Status

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/client.md

Retrieve the status of a single workflow by its unique ID. Returns the WorkflowStatus object if found, otherwise None.

```python
status = client.get_workflow("order_123")
if status:
    print(f"Status: {status.status}, Result: {status.output}")
```

--------------------------------

### Define an Async Workflow

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/utilities.md

This example demonstrates how to define an asynchronous workflow in DBOS, which utilizes the managed async event loop for executing asynchronous operations.

```python
@DBOS.workflow()
async def async_workflow():
    result = await some_async_step()
    return result
```

--------------------------------

### AsyncSQLAlchemyDatasource Constructor

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/datasource.md

Initializes an instance of the AsyncSQLAlchemyDatasource. It can be configured with a name, a configuration name, an existing async engine, a database URL, or engine keyword arguments.

```APIDOC
## AsyncSQLAlchemyDatasource Constructor

### Description
Creates an async datasource.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **name** (`str`) - Required - Instance name
- **config_name** (`Optional[str]`) - Optional - Configuration name
- **engine** (`Optional[sa.ext.asyncio.AsyncEngine]`) - Optional - Existing async engine
- **url** (`Optional[str]`) - Optional - Database URL
- **engine_kwargs** (`Optional[dict[str, Any]]`) - Optional - Engine options

### Request Example
```python
async_ds = AsyncSQLAlchemyDatasource(
    name="async_db",
    url="postgresql+asyncpg://user:pass@localhost/myapp",
)
```

### Response
#### Success Response (200)
None

#### Response Example
None
```

--------------------------------

### Scheduled Workflow Execution

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/README.md

Schedule workflows to run automatically at specified intervals using cron syntax. The example shows an hourly cleanup workflow.

```python
@DBOS.scheduled("0 * * * *")  # Hourly
@DBOS.workflow()
def hourly_cleanup(scheduled_time, actual_time):
    delete_old_logs()
```

--------------------------------

### Get Asynchronous SQLAlchemy Session

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Retrieve an asynchronous SQLAlchemy session for the application database using `get_async_session`. This is suitable for use in async DBOS operations.

```python
async_session = DBOS.get_async_session()
user = await async_session.get(User, user_id)
```

--------------------------------

### Get Executor ID

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/context.md

Retrieve the unique identifier for the executor running the current workflow. This is useful for logging or conditional logic based on the execution environment.

```python
ctx = DBOS.get_context()
print(f"Running on executor: {ctx.executor_id}")
```

--------------------------------

### Single Process Deployment

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/README.md

Launches the DBOS application in a single process. This is suitable for development or simple deployments. Ensure DBOSConfig is correctly set up via environment variables.

```python
from dbos import DBOS, DBOSConfig

config = DBOSConfig.from_env()
dbos = DBOS(config=config)
dbos.launch()

# Server code here
# Workflows start as needed
```

--------------------------------

### StepOptions

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/types.md

Configuration options for step decorators in DBOS. This TypedDict allows customization of step names, maximum retries, and execution timeouts.

```APIDOC
## StepOptions

### Description
Configuration options for step decorators in DBOS. This TypedDict allows customization of step names, maximum retries, and execution timeouts.

### Fields
- **name** (`Optional[str]`, Optional): Custom name for the step; defaults to function name
- **max_retries** (`int`, Required): Maximum number of times to retry on failure
- **timeout_seconds** (`Optional[float]`, Optional): Maximum execution time in seconds

### Module
`dbos._core`

### Example
```python
@DBOS.step(name="charge_card", max_retries=3, timeout_seconds=30)
def charge_payment(amount: float):
    return payment_api.charge(amount)
```
```

--------------------------------

### Start Workflow with Exactly-Once Semantics

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/README.md

Use SetWorkflowID to ensure a workflow is executed only once, even if called multiple times with the same request ID. This is crucial for idempotent operations.

```python
with SetWorkflowID(request_id):
    DBOS.start_workflow(my_workflow)
# If called again with same request_id, returns existing result
```

--------------------------------

### PostgreSQL Connection with SSL

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Configure DBOS to connect to a PostgreSQL database using SSL.

```python
config = DBOSConfig(
    database_url="postgresql://user:pass@host/db?sslmode=require",
)
```

--------------------------------

### Durable Sleep Example

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/README.md

Shows how to use DBOS.sleep to pause a workflow for a specified duration, ensuring it resumes after interruptions.

```python
@DBOS.workflow()
def reminder_workflow(email: str, time_to_sleep: int):
    send_confirmation_email(email)
    DBOS.sleep(time_to_sleep)
    send_reminder_email(email)
```

--------------------------------

### DBOS Constructor

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Creates or retrieves the singleton DBOS instance. DBOS uses a singleton pattern—multiple calls with the same arguments return the same instance.

```APIDOC
## DBOS Constructor

### Description
Creates or retrieves the singleton DBOS instance. DBOS uses a singleton pattern—multiple calls with the same arguments return the same instance.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **config** (`DBOSConfig`) - Required - Configuration object with database and runtime settings
- **fastapi** (`Optional[FastAPI]`) - Optional - FastAPI application to integrate DBOS middleware
- **flask** (`Optional[Flask]`) - Optional - Flask application to integrate DBOS middleware
- **conductor_url** (`Optional[str]`) - Optional - URL of a Conductor service for multi-machine orchestration
- **conductor_key** (`Optional[str]`) - Optional - API key for Conductor authentication

### Request Example
```python
from dbos import DBOS, DBOSConfig

config = DBOSConfig.from_env()
dbos = DBOS(config=config)
dbos.launch()
```

### Response
#### Success Response (200)
- **DBOS Instance** (`DBOS`) - The singleton DBOS instance.

#### Response Example
```python
# No specific example for the return value, but the constructor call is shown above.
```
```

--------------------------------

### SQLite In-Memory Database for Testing

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Configure DBOS to use an in-memory SQLite database, suitable for testing.

```python
config = DBOSConfig(
    database_url="sqlite:///:memory:",
)
```

--------------------------------

### Configure Separate System and Application Databases

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Specify distinct database URLs for DBOS system tables and application-specific data.

```python
config = DBOSConfig(
    system_database_url="postgresql://user:pass@localhost/dbos_system",
    application_database_url="postgresql://user:pass@localhost/dbos_app",
)
```

--------------------------------

### Get DBOS Execution Context

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Retrieve the current DBOS execution context using `get_context`. This context provides access to workflow state, authentication, and logging.

```python
ctx = DBOS.get_context()
current_user = ctx.authenticated_user
logger = ctx.logger
```

--------------------------------

### DBOSClient Constructor

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/client.md

Initializes a DBOSClient for accessing DBOS workflow data. This client is for read and write operations and does not launch DBOS or run migrations.

```APIDOC
## DBOSClient Constructor

### Description
Initializes a client for accessing DBOS workflow data. This client is intended for read and write operations and does not launch DBOS or run migrations.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **database_url** (`Optional[str]`) - Optional - Deprecated; use application_database_url
- **system_database_url** (`Optional[str]`) - Optional - PostgreSQL/SQLite URL for DBOS system tables
- **system_database_engine** (`Optional[sa.Engine]`) - Optional - SQLAlchemy engine for system database
- **application_database_url** (`Optional[str]`) - Optional - PostgreSQL URL for application data
- **dbos_system_schema** (`Optional[str]`) - Optional - Schema name for DBOS system tables (Default: "dbos")
- **serializer** (`Serializer`) - Optional - Custom serializer for workflow args/results (Default: DefaultSerializer())

### Request Example
```python
client = DBOSClient(
    system_database_url="postgresql://user:pass@localhost/dbos",
    application_database_url="postgresql://user:pass@localhost/app",
)
```

### Response
#### Success Response (None)
None

#### Response Example
None
```

--------------------------------

### Configure Custom Serializer

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Integrate a custom serializer for workflow arguments and results by implementing the `Serializer` protocol.

```python
from dbos import DBOSConfig, DBOSPortableJSONSerializer

config = DBOSConfig(
    database_url="...",
    serializer=DBOSPortableJSONSerializer(),
)
```

--------------------------------

### Get DBOS Executor/Machine ID

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Obtain the executor or machine ID set by DBOS Cloud using the DBOS__VMID environment variable. Defaults to 'local' if not specified.

```python
import os

executor_id = os.environ.get("DBOS__VMID", "local")
```

--------------------------------

### Initialize DBOSConfig with Portable JSON Serializer

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Configure DBOSConfig to use the Portable JSON Serializer for data serialization.

```python
from dbos import DBOSPortableJSONSerializer

config = DBOSConfig(
    database_url="...",
    serializer=DBOSPortableJSONSerializer(),
)
```

--------------------------------

### Get DBOS Application ID

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Retrieve the application ID assigned by DBOS Cloud using the DBOS__APPID environment variable. Defaults to an empty string if not set.

```python
import os

app_id = os.environ.get("DBOS__APPID", "")
```

--------------------------------

### PostgreSQL Connection Parameters

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/datasource.md

Append connection options to PostgreSQL URLs to configure SSL mode, timeouts, and connection pooling.

```sql
postgresql://user:password@host/db?sslmode=require&timeout=10
```

--------------------------------

### Enqueue Workflow Synchronously

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/queue.md

Enqueues a workflow to run asynchronously and returns a handle to poll for completion. Use when you need to start a workflow and potentially wait for its result later.

```python
handle = queue.enqueue(send_email, "user@example.com", subject="Welcome")
result = handle.get_result()  # Wait for completion
```

--------------------------------

### Type-Aware Deserialization

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/utilities.md

Implement a deserializer that uses type hints to correctly parse incoming byte data. This example shows basic JSON parsing based on type.

```python
import json

class TypeAwareSerializer:
    def deserialize(self, data: bytes, type_hint: Optional[type]) -> Any:
        if type_hint == list:
            return json.loads(data)
        elif type_hint == dict:
            return json.loads(data)
        else:
            return json.loads(data)
```

--------------------------------

### Run unit tests

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/DEVELOPING.md

Executes unit tests for the project using pytest.

```bash
pdm run pytest tests
```

--------------------------------

### run_dbos_database_migrations()

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/utilities.md

Runs all pending DBOS migrations on the database using the provided DBOSConfig.

```APIDOC
## run_dbos_database_migrations()

### Description
Runs all pending DBOS migrations on the database.

### Module
`dbos.cli.migration`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

#### Parameters
- **config** (`DBOSConfig`) - Required - Configuration with database URL

### Request Example
```python
from dbos import run_dbos_database_migrations, DBOSConfig

config = DBOSConfig(database_url="postgresql://...")
run_dbos_database_migrations(config)
```

### Response
None

### Example Usage in an Async Main Function
```python
import asyncio
from dbos import DBOSConfig, run_dbos_database_migrations, DBOS

async def main():
    config = DBOSConfig.from_env()
    
    # Run migrations
    run_dbos_database_migrations(config)
    
    # Launch DBOS
    dbos = DBOS(config=config)
    dbos.launch()
    
    # Run workflows
    ...

asyncio.run(main())
```
```

--------------------------------

### Get Synchronous SQLAlchemy Session

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/dbos-core.md

Obtain a SQLAlchemy synchronous session for the application database using `get_session`. This must be called within a DBOS workflow, step, or transaction.

```python
session = DBOS.get_session()
user = session.query(User).filter_by(id=user_id).one()
```

--------------------------------

### Unit Test with In-Memory SQLite Database

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/README.md

This snippet demonstrates how to set up a unit test using an in-memory SQLite database. It uses a pytest fixture to initialize and tear down the DBOS instance with an in-memory database.

```python
import pytest
from dbos import DBOSConfig, DBOS

@pytest.fixture
def dbos():
    config = DBOSConfig(database_url="sqlite:///:memory:")
    instance = DBOS(config=config)
    instance.launch()
    yield instance
    instance.destroy(destroy_registry=True)

def test_workflow(dbos):
    handle = dbos.start_workflow(my_workflow, "param")
    result = handle.get_result()
    assert result == expected
```

--------------------------------

### Access Configuration Value usinggetitem

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Use dictionary-like bracket notation to access a configuration value by its key.

```python
version = config["application_version"]
```

--------------------------------

### Access Step Status Information

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/context.md

Get status details for the current step execution, including its ID and the number of attempts. This can be used for monitoring or implementing retry logic.

```python
ctx = DBOS.get_context()
if ctx.step_status:
    print(f"Step {ctx.step_status.step_id}, attempt {ctx.step_status.current_attempt}")
```

--------------------------------

### SetWorkflowID Context Manager

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/context.md

Allows assigning a specific ID to the next workflow started within its context, facilitating idempotent workflow execution by ensuring a workflow with the same ID is not re-executed.

```APIDOC
## SetWorkflowID Context Manager

Assigns a specific ID to the next workflow started within the context block. Enables idempotent workflow execution.

**Module:** `dbos._context`

```python
class SetWorkflowID:
    def __init__(self, workflow_id: str) -> None: ...
    def __enter__(self) -> None: ...
    def __exit__(self, exc_type, exc_val, exc_tb) -> None: ...
```

| Parameter | Type | Description |
|-----------|------|-------------|
| workflow_id | `str` | The ID to assign to the workflow |

**Example:**

```python
request_id = request.headers["X-Request-ID"]
with SetWorkflowID(request_id):
    handle = DBOS.start_workflow(process_request, request)
    return {"workflow_id": handle.get_workflow_id()}
```
```

--------------------------------

### Implement Custom JSON Serializer

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/types.md

Example of a custom serializer using Python's built-in json module. This custom serializer can be provided to DBOSConfig for specific serialization needs.

```python
class CustomSerializer:
    def serialize(self, obj: Any) -> bytes:
        return json.dumps(obj).encode()
    
    def deserialize(self, data: bytes, type_hint: Optional[type]) -> Any:
        return json.loads(data.decode())

config = DBOSConfig(
    database_url="...",
    serializer=CustomSerializer(),
)
```

--------------------------------

### Set SQLite Database URL

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/configuration.md

Configure a SQLite database file path for DBOS.

```python
config = DBOSConfig(
    database_url="sqlite:///path/to/database.db"
)
```

--------------------------------

### Processing Kafka Messages in a Workflow

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/types.md

Example of a DBOS workflow that consumes Kafka messages. It demonstrates how to access message fields like topic and value, and parse the JSON payload.

```python
@DBOS.kafka_consumer(kafka_config, ["orders"])
@DBOS.workflow()
def process_order(msg: KafkaMessage):
    order_data = json.loads(msg.value.decode())
    print(f"Processing order from topic {msg.topic}")
```

--------------------------------

### Queue Constructor

Source: https://github.com/dbos-inc/dbos-transact-py/blob/main/_autodocs/api-reference/queue.md

Creates a queue for managing workflow execution. You can configure concurrency, rate limiting, priority, partitioning, and other operational parameters.

```APIDOC
## Queue Constructor

### Description
Creates a queue for managing workflow execution. You can configure concurrency, rate limiting, priority, partitioning, and other operational parameters.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Method Signature
```python
Queue(
    name: str,
    concurrency: Optional[int] = None,
    limiter: Optional[QueueRateLimit] = None,
    *, 
    worker_concurrency: Optional[int] = None,
    priority_enabled: bool = False,
    partition_queue: bool = False,
    polling_interval_sec: float = 1.0,
    database_backed_queue: bool = False,
    client_system_database: Optional[SystemDatabase] = None,
) -> None
```

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| name | `str` | — | Unique queue name |
| concurrency | `Optional[int]` | None | Maximum concurrent workflows (unlimited if None) |
| limiter | `Optional[QueueRateLimit]` | None | Rate limit configuration |
| worker_concurrency | `Optional[int]` | None | Concurrency per individual worker |
| priority_enabled | `bool` | False | Enable priority-based task execution |
| partition_queue | `bool` | False | Enable partition-based queue routing |
| polling_interval_sec | `float` | 1.0 | Poll interval for dequeuing tasks |
| database_backed_queue | `bool` | False | Store queue config in database instead of memory |
| client_system_database | `Optional[SystemDatabase]` | None | SystemDatabase for client-side operations |

### Throws
| Error | Condition |
|---|---|
| `ValueError` | Invalid concurrency or polling configuration |
| `Exception` | Queue name already registered |
```