aiogram-dependency 1.1.0

A FastAPI-style dependency injection system for aiogram Telegram bots. This library brings clean, type-safe dependency injection to your aiogram handlers, making your code more modular, testable, and maintainable.

Aiogram Dependency Injection

A FastAPI-style dependency injection system for aiogram Telegram bots. This library brings clean, type-safe dependency injection to your aiogram handlers, making your code more modular, testable, and maintainable.

Features

• FastAPI-style syntax - Familiar Depends() decorator
• Multiple dependency scopes - Singleton, Request, and Transient
• Nested dependencies - Dependencies can depend on other dependencies
• Async support - Both sync and async dependency functions
• Circular dependency detection - Prevents infinite loops
• Type-safe - Full type hints support
• Smart caching - Efficient resource management
• FastAPI Support - Can use fastapi Depends instead default one

Installation

pip install aiogram-dependency

Or install from source:

git clone https://github.com/AstralMortem/aiogram-dependency
cd aiogram-dependency
pip install -e .

Quick Start

import asyncio
from aiogram import Bot, Dispatcher, F
from aiogram.types import Message

from aiogram_dependency import Depends, setup_dependency, Scope

# Define your dependencies
class DatabaseConnection:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
    
    async def query(self, sql: str):
        # Your database logic here
        return f"Result for: {sql}"

class UserService:
    def __init__(self, db: DatabaseConnection):
        self.db = db
    
    async def get_user_profile(self, user_id: int):
        return await self.db.query(f"SELECT * FROM users WHERE id = {user_id}")

# Dependency factories
async def get_database() -> DatabaseConnection:
    return DatabaseConnection("postgresql://localhost/mydb")

async def get_user_service(
    db: DatabaseConnection = Depends(get_database, scope=Scope.SINGLETON)
) -> UserService:
    return UserService(db)

# Handler with dependency injection
async def profile_handler(
    message: Message,
    user_service: UserService = Depends(get_user_service)
):
    if not message.from_user:
        await message.answer("User not found")
        return
    
    profile = await user_service.get_user_profile(message.from_user.id)
    await message.answer(f"Your profile: {profile}")

# Setup bot (ORDER MATER!)
async def main():
    bot = Bot(token="YOUR_BOT_TOKEN")
    dp = Dispatcher()
    
    # Register handlers
    dp.message.register(profile_handler, F.text == "/profile")

    # Register dependency injection 
    setup_dependency(dp)
    
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

Dependency Scopes

Singleton

Created once and shared across all requests globally.

async def get_database() -> DatabaseConnection:
    return DatabaseConnection("connection_string")

async def handler(
    message: Message,
    db: DatabaseConnection = Depends(get_database, scope=Scope.SINGLETON)
):
    # Same database instance for all users
    pass

Request (Default)

Created once per user/chat and cached for subsequent calls in the same context.

async def get_user_service() -> UserService:
    return UserService()

async def handler(
    message: Message,
    service: UserService = Depends(get_user_service, scope=Scope.REQUEST)
):
    # Same service instance for this user, different for other users
    pass

Transient

Created fresh every time it's requested.

async def get_timestamp() -> float:
    return time.time()

async def handler(
    message: Message,
    timestamp: float = Depends(get_timestamp, scope=Scope.TRANSIENT)
):
    # New timestamp every time
    pass

Advanced Usage

FastAPI

You can use FastAPI Depends to remove dupplication if you use fastapi as backend server

Annotated

You can use Annotated type to make it more readable.

from typing import Annotated
from aiogram_dependency import Depends
from aiogram import Dispatcher,filters,types

async def get_database() -> DatabaseConnection:
    return DatabaseConnection("postgresql://localhost/db")

DatabaseDep = Annotated[DatabaseConnection, Depends(get_database)]

async def get_user(db: DatabaseDep):
    return db.execute('SQL')

UserDep = Annotated[dict, Depends(get_user)]

dp = Dispatcher()

@dp.message(filters.CommandStart())
async def start(message:types.Message, user: UserDep):
    return await message.answer(user['username'])

Nested Dependencies

Dependencies can depend on other dependencies:

async def get_database() -> DatabaseConnection:
    return DatabaseConnection("postgresql://localhost/db")

async def get_user_repository(
    db: DatabaseConnection = Depends(get_database)
) -> UserRepository:
    return UserRepository(db)

async def get_user_service(
    user_repo: UserRepository = Depends(get_user_repository),
    notification_service: NotificationService = Depends(get_notification_service)
) -> UserService:
    return UserService(user_repo, notification_service)

Using Event Data in Dependencies

Dependencies can access the current event and handler data:

async def get_current_user(event: Message) -> Optional[User]:
    return event.from_user

async def get_user_permissions(
    event: Message,
    data: dict,
    current_user: User = Depends(get_current_user)
) -> List[str]:
    # Access event, data, and other dependencies
    if current_user and current_user.id in data.get('admins', []):
        return ['admin', 'user']
    return ['user']

async def admin_handler(
    message: Message,
    permissions: List[str] = Depends(get_user_permissions)
):
    if 'admin' not in permissions:
        await message.answer("Access denied")
        return
    
    await message.answer("Welcome, admin!")

Custom Registry and Resolver

For advanced use cases, you can customize the dependency system:

from aiogram_dependency.registry import DependencyRegistry
from aiogram_dependency.resolver import DependencyResolver
from aiogram_dependency.middleware import DependencyMiddleware
from aiogram_dependency import Scope

# Create custom registry
registry = DependencyRegistry()

db_dep = Depends(get_database, scope=Scope.SINGLETON)

# Pre-populate with some dependencies
registry.set_dependency(
    db_dep,
    DatabaseConnection("custom://connection"),
    "global"
)

# Use custom middleware
middleware = DependencyMiddleware(registry)
dp.message.middleware(middleware)

Testing

The library is fully testable. Here's an example:

Run the full test suite:

# Install test dependencies
pip install pytest pytest-asyncio

# Run tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=aiogram_dependency --cov-report=html

License

This project is licensed under the MIT License - see the LICENSE file for details.

Related Projects

• aiogram - Modern and fully asynchronous framework for Telegram Bot API
• FastAPI - Modern, fast web framework for building APIs with Python