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.
- 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
pip install aiogram-dependencyOr install from source:
git clone https://github.com/AstralMortem/aiogram-dependency
cd aiogram-dependency
pip install -e .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())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
passCreated 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
passCreated 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
passYou can use FastAPI Depends to remove dupplication if you use fastapi as backend server
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'])
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)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!")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)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=htmlThis project is licensed under the MIT License - see the LICENSE file for details.