FastAPI dependencies¶
When using aiosend.webhook.FastAPIManager, you can use
FastAPI dependencies
directly inside aiosend event handlers.
This allows you to reuse the same dependencies from your FastAPI application, for example database sessions, services, authentication data or request context.
Tip
To use aiosend with FastAPI, install the fastapi extra:
pip install aiosend[fastapi]
Usage example¶
import asyncio
from typing import Annotated
import uvicorn
from fastapi import Depends, FastAPI, Request
from aiosend import CryptoPay
from aiosend.types import Invoice
from aiosend.webhook import FastAPIManager
app = FastAPI(title="My App")
cp = CryptoPay("TOKEN", webhook_manager=FastAPIManager(app, "/handler"))
async def get_app_name(request: Request) -> str:
return request.app.title
@cp.invoice_paid()
async def handler(
invoice: Invoice,
app_name: Annotated[str, Depends(get_app_name)],
) -> None:
print(f"Received {invoice.amount} {invoice.asset} in {app_name}")
async def main() -> None:
invoice = await cp.create_invoice(1, "USDT")
print("invoice link:", invoice.bot_invoice_url)
if __name__ == "__main__":
asyncio.run(main())
uvicorn.run(app)
How it works¶
When a webhook update is received, aiosend.webhook.FastAPIManager
uses FastAPI’s dependency injection system to resolve dependencies declared
in the matched aiosend event handler.
The event object itself, such as aiosend.types.Invoice, is provided
by aiosend:
async def get_service() -> Service:
return Service()
@cp.invoice_paid()
async def handler(
invoice: Invoice,
service: Annotated[Service, Depends(get_service)],
) -> None:
await service.process(invoice)
Webhook route dependencies¶
aiosend.webhook.FastAPIManager also accepts FastAPI dependencies
that should be executed for the webhook route itself:
async def verify_webhook() -> None:
...
cp = CryptoPay(
"TOKEN",
webhook_manager=FastAPIManager(
app,
"/handler",
dependencies=[
Depends(verify_webhook),
],
),
)
Nested dependencies¶
Dependencies can depend on other FastAPI dependencies as usual:
async def get_database() -> Database:
return Database()
async def get_service(
database: Annotated[Database, Depends(get_database)],
) -> Service:
return Service(database)
@cp.invoice_paid()
async def handler(
invoice: Invoice,
service: Annotated[Service, Depends(get_service)],
) -> None:
await service.process(invoice)