Dependencies

Dependencies let you share common logic across channel handlers. They are declared with Depends and injected using typing.Annotated.

A dependency function can be def or async def. It can also be a generator (sync or async) to provide setup and cleanup logic.

Basic Dependency

Use Depends inside Annotated to tell AsyncFast to resolve a value and pass it to your handler:

Example

from typing import Annotated

from asyncfast import AsyncFast
from asyncfast import Depends
from asyncfast import Header

app = AsyncFast()


def get_context(
    request_id: Annotated[str, Header(alias="request-id")],
) -> dict[str, str]:
    return {"request_id": request_id}


@app.channel("orders.created")
async def handle_orders(
    context: Annotated[dict[str, str], Depends(get_context)],
) -> None:
    print(context["request_id"])

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "HandleOrders": {
      "address": "orders.created",
      "messages": {
        "HandleOrdersMessage": {
          "$ref": "#/components/messages/HandleOrdersMessage"
        }
      }
    }
  },
  "operations": {
    "receiveHandleOrders": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/HandleOrders"
      }
    }
  },
  "components": {
    "messages": {
      "HandleOrdersMessage": {
        "headers": {
          "$ref": "#/components/schemas/HandleOrdersMessageHeaders"
        }
      }
    },
    "schemas": {
      "HandleOrdersMessageHeaders": {
        "properties": {
          "request-id": {
            "title": "Request-Id",
            "type": "string"
          }
        },
        "required": [
          "request-id"
        ],
        "title": "HandleOrdersMessageHeaders",
        "type": "object"
      }
    }
  }
}

Dependencies Use The Same Inputs

Dependency functions can declare the same kinds of parameters as a channel handler:

• Payload (the message body)

• Header values

• channel parameters (from the address template)

• bindings (via protocol binding types)

• MessageSender for sending follow-up messages

These inputs are resolved for dependencies exactly the same way they are for handlers.

Sub-dependencies

Dependencies can depend on other dependencies using the same Depends pattern:

Example

from typing import Annotated

from asyncfast import AsyncFast
from asyncfast import Depends
from asyncfast import Header

app = AsyncFast()


def get_tenant_id(tenant_id: Annotated[str, Header(alias="tenant-id")]) -> str:
    return tenant_id


def get_context(
    tenant_id: Annotated[str, Depends(get_tenant_id)],
) -> dict[str, str]:
    return {"tenant_id": tenant_id}


@app.channel("billing")
async def handle_billing(
    context: Annotated[dict[str, str], Depends(get_context)],
) -> None:
    print(context["tenant_id"])

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "HandleBilling": {
      "address": "billing",
      "messages": {
        "HandleBillingMessage": {
          "$ref": "#/components/messages/HandleBillingMessage"
        }
      }
    }
  },
  "operations": {
    "receiveHandleBilling": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/HandleBilling"
      }
    }
  },
  "components": {
    "messages": {
      "HandleBillingMessage": {
        "headers": {
          "$ref": "#/components/schemas/HandleBillingMessageHeaders"
        }
      }
    },
    "schemas": {
      "HandleBillingMessageHeaders": {
        "properties": {
          "tenant-id": {
            "title": "Tenant-Id",
            "type": "string"
          }
        },
        "required": [
          "tenant-id"
        ],
        "title": "HandleBillingMessageHeaders",
        "type": "object"
      }
    }
  }
}

Cleanup With Yield

If a dependency is a generator, AsyncFast treats it like a context manager and runs cleanup after the handler returns. This works for both def generators and async def generators:

Example

from collections.abc import AsyncGenerator
from typing import Annotated

from asyncfast import AsyncFast
from asyncfast import Depends

app = AsyncFast()


async def get_resource() -> AsyncGenerator[str, None]:
    resource = "connected"
    try:
        yield resource
    finally:
        # Cleanup happens after the handler returns.
        pass


@app.channel("ping")
async def handle_ping(
    resource: Annotated[str, Depends(get_resource)],
) -> None:
    print(resource)

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "HandlePing": {
      "address": "ping",
      "messages": {
        "HandlePingMessage": {
          "$ref": "#/components/messages/HandlePingMessage"
        }
      }
    }
  },
  "operations": {
    "receiveHandlePing": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/HandlePing"
      }
    }
  },
  "components": {
    "messages": {
      "HandlePingMessage": {}
    }
  }
}

Caching

By default, a dependency result is cached per message and reused if requested multiple times. To disable caching, set use_cache=False on Depends:

Example

from typing import Annotated

from asyncfast import AsyncFast
from asyncfast import Depends
from asyncfast import Header

app = AsyncFast()


def get_request_id(request_id: Annotated[str, Header(alias="request-id")]) -> str:
    return request_id


@app.channel("events")
async def handle_events(
    request_id: Annotated[str, Depends(get_request_id, use_cache=False)],
    request_id_again: Annotated[str, Depends(get_request_id, use_cache=False)],
) -> None:
    print(request_id, request_id_again)

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "HandleEvents": {
      "address": "events",
      "messages": {
        "HandleEventsMessage": {
          "$ref": "#/components/messages/HandleEventsMessage"
        }
      }
    }
  },
  "operations": {
    "receiveHandleEvents": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/HandleEvents"
      }
    }
  },
  "components": {
    "messages": {
      "HandleEventsMessage": {
        "headers": {
          "$ref": "#/components/schemas/HandleEventsMessageHeaders"
        }
      }
    },
    "schemas": {
      "HandleEventsMessageHeaders": {
        "properties": {
          "request-id": {
            "title": "Request-Id",
            "type": "string"
          }
        },
        "required": [
          "request-id"
        ],
        "title": "HandleEventsMessageHeaders",
        "type": "object"
      }
    }
  }
}