Async-FastAPI-MultiDB

English | 中文

Introduction

This project is built on a modern and cutting-edge technology stack with a clear layered architecture design, aiming to deliver a high-performance, scalable, and secure backend solution. By strictly adhering to the Google Python Style Guide, integrating type checking, and automated testing, it ensures code quality and stability.

The project supports multiple databases (both relational and NoSQL) and fully embraces asynchronous programming to enhance response efficiency. Leveraging FastAPI’s powerful automatic API documentation features facilitates development and maintenance. On the security front, it employs JWT + RSA authentication combined with RBAC-based fine-grained access control and Redis caching for efficient permission verification.

Additionally, comprehensive environment configuration management, multi-environment support, and detailed code annotations greatly improve development and operational experience. This project is well-suited for building modern distributed systems and microservice architectures.

Core technologies include FastAPI, Socket.IO, Celery, MinIO, SQLModel, Beanie, among others, to meet diverse business needs.

---

Table of Contents

• Project Features
• Quick Start
• Architecture
• Structure
• Authentication & Authorization
• Websocket
• Celery Async Task
• Test
• License

Features

• Cutting-edge Technology Stack: Utilizes modern and efficient technologies such as FastAPI, Socket.IO, Celery, MinIO, SQLModel, and Beanie to build a state-of-the-art backend solution.

• Clear Project Architecture: Based on a Layered Architecture, with well-defined responsibilities and clear separation of concerns, enhancing maintainability and scalability.

• Strict Code Standards: Fully adheres to the Google Python Style Guide and integrates tools like ruff, mypy, and pre-commit to ensure consistent coding style and type safety.

• Automated API Documentation: Leverages FastAPI's built-in features to automatically generate interactive API docs, supporting Swagger UI and Redoc for easy development and testing.

• Comprehensive Authentication and Authorization System: Implements secure authentication using JWT combined with RSA, with fine-grained role-based access control (RBAC) and Redis-backed permission caching for high-performance validation.

• Detailed Code Comments and Type Annotations: Core modules are thoroughly documented with clear comments and strict typing, improving code readability and team collaboration.

• Environment Configuration Management: Supports multi-environment configuration via .env files for flexible management of development, testing, and production settings.

• Multi-database Support: Seamlessly supports SQLModel (based on SQLAlchemy), SQLAlchemy, and Beanie (MongoDB ODM), catering to diverse data storage needs.

• High-performance Asynchronous Architecture: Fully embraces async/await asynchronous programming patterns to enhance system throughput and responsiveness.

• Comprehensive Testing Framework: Integrates Pytest with full test coverage to ensure system stability and reliability.

🚧 This project is under active development. Feel free to follow, star the repo, or contribute via issues and PRs.

---

Quick Start

1. Clone the repository:

   git clone https://github.com/GJCoke/Async-FastAPI-MultiDB.git
   cd Async-FastAPI-MultiDB

2. Copy the environment variables:

   cp .env.example .env

3. Run with Docker:

   docker network create app_network
   docker compose up -d --build

4. Run Alembic to create the database schema.

   docker compose exec app scripts/alembic/makemigrations.sh "Initialize Database"
   docker compose exec app scripts/alembic/migrate.sh

5. Run the initdb script to generate the necessary initial data.

   docker compose exec app scripts/initdb.sh

6. Default Username and Password

   After initialization, a default user will be created with the following credentials:

   • Username：admin
   • Password：123456

   These credentials can be used to access the system or for debugging authentication-related endpoints.

   ⚠️ Please make sure to change the default password promptly in production environments!

7. Development workflow: This project uses pre-commit to enforce code quality and consistency:

   pre-commit install

   The .pre-commit-config.yaml includes:

   • Large file check: Whether large files have been added to the committed code.
   • Auto formatting via ruff
   • Static type checking with mypy

Access the Swagger UI at: http://localhost:16000/docs

Error responses are globally enhanced—no need to define them on each route individually.

---

Architecture

This document provides a basic architectural overview of a FastAPI project, aiming to help developers understand the project's organization and the functionality of each module. Through a detailed analysis of the project structure, readers can quickly grasp how to build and maintain an efficient FastAPI application.

Structure Description

src/
│
├── api/                  # API route definitions, organized by version
│   ├── v1/               # v1 version APIs
│   │   ├── auth.py       # Login, registration, and permission-related APIs
│   │   └── router.py     # v1 route aggregation
│   └── v2/               # Reserved or under development for v2
│       └── __init__.py
│
├── core/                 # Core functionality modules
│   ├── config.py         # Load environment variables and manage Settings
│   ├── database.py       # Database connections (SQLModel, Mongo, etc.)
│   ├── environment.py    # Environment detection (e.g., dev/test/prod)
│   ├── exceptions.py     # Custom exception classes
│   ├── lifecycle.py      # FastAPI application lifecycle events
│   └── route.py          # Support for dynamic route registration
│
├── crud/                 # Direct CRUD operations for the database
│   └── router.py         # Example or common DB operations
│
├── deps/                 # FastAPI dependencies (used with Depends)
│   ├── auth.py           # Permission/role validation dependencies
│   ├── database.py       # DB connection dependencies
│   ├── environment.py    # Environment-related dependencies
│   ├── role.py           # Role-based permission injection
│   └── router.py         # Route-level dependencies
│
├── middlewares/          # middlewares
│   ├── logger.py         # http logging
│
├── models/               # Database model definitions
│   ├── auth.py           # Tables for users, permissions, etc.
│   ├── base.py           # Common base classes (timestamps, ID, etc.)
│   └── router.py         # Route model definitions (e.g., permission routes)
│
├── queues/               # Celery async task modules
│   ├── tasks/            # Task definitions
│   │   └── tasks.py      # Sample async task collection
│   ├── app.py            # Celery instance creation
│   ├── celery.py         # Celery startup entry point
│   ├── models.py         # Models related to async tasks (e.g., task logs)
│   ├── scheduler.py      # Scheduled task scheduler
│   └── task.py           # Task registration and encapsulation
│
├── schemas/              # Request/response data structure definitions
│   ├── auth.py           # Models for login, registration, etc.
│   ├── base.py           # Common field models
│   ├── request.py        # Request data models
│   ├── response.py       # Response models (standard format)
│   ├── role.py           # Role-related schemas
│   └── router.py         # Route/API-related schemas
│
├── services/             # Business logic layer
│   └── auth.py           # User authentication services (login validation, token generation, etc.)
│
├── utils/                # Utility method collections
│   ├── constants.py      # Global constant definitions
│   ├── date.py           # Date/time handling functions
│   ├── minio_client.py   # MinIO object storage wrapper
│   ├── security.py       # Encryption and JWT utilities
│   ├── uuid7.py          # Custom UUID utilities
│   └── validate.py       # Field/form validation utilities
│
├── websockets/           # WebSocket routes and logic
│   ├── __init__.py
│   ├── events/           # WebSocket events (connect, disconnect, join room, etc.)
│   ├── server.py         # Socket.IO wrapper with added Pydantic integration
│   ├── app.py            # WebSocket application entry point
│
├── initdb.py             # Database initialization script (e.g., table creation, insert default data)
├── main.py               # FastAPI application entry point

---

Auth Module Overview

This module handles authentication and authorization, built upon JWT + Redis + RSA + RBAC.

Features Overview

• User login via username and password
• AccessToken / RefreshToken generation and validation
• Token refresh
• Token logout
• Encapsulated user info injection via dependency
• Environment-based restrictions (e.g., Debug-only features)
• RBAC access control model

Password Encryption (RSA)

You don't need to worry about Swagger UI being affected by RSA encryption — it uses an independent login flow, which only works in the DEBUG environment.

During login, the frontend encrypts the password using the RSA public key provided by the backend. The backend then decrypts it using the private key, ensuring that the password is never transmitted in plain text.

It is recommended to configure the key pair via environment variables.

The DEBUG environment supports dynamic key generation, but this is not recommended in production.

Dynamic keys can lead to inconsistent behavior across multiple services or instances, especially in environments with load balancing or distributed caching (e.g., Redis).

Token Description

• AccessToken: Short-lived, stored on the client, used for authenticating API requests
• RefreshToken: Long-lived, stored in Redis, used to refresh access tokens

The RefreshToken embeds a unique jti (JWT ID) and User-Agent to ensure the refresh request originates from the same source.

RBAC

• No need to manually define permission codes for each API endpoint; the system automatically generates and matches permission identifiers based on the endpoint path and HTTP method, achieving fully automated permission management.
• User permission data is stored in Redis as structured data and cached according to the access-token lifecycle, effectively reducing database lookups and greatly improving permission-check efficiency and overall system response speed.

Core Dependencies

名称	说明
HeaderAccessTokenDep	Extracts the AccessToken from the request header
HeaderRefreshTokenDep	Extracts the RefreshToken from the request header
HeaderUserAgentDep	Extracts the User-Agent from the request header
UserAccessJWTDep	Decodes the AccessToken and retrieves user information
UserRefreshJWTDep	Decodes the RefreshToken and validates the User-Agent
AuthCrudDep	CRUD wrapper for user-related database operations
UserRefreshDep	Validates and retrieves user info via Redis + Database
UserDBDep	Retrieves user information directly from the database
VerifyPermissionDep	Route-based User Access Permission Validation

Route Summary

• GET /keys/public: Retrieve the RSA public key used for encrypting passwords
• POST /login: User login, returns both access_token and refresh_token
• POST /token/refresh: Refresh the token, requires refresh_token and User-Agent
• POST /logout: Logout, deletes the refresh token from Redis
• GET /user/info: Retrieve current user information
• GET /router/backend：Get Application Routes

Redis Structure

• Stored Key: auth:refresh:<{user_id}>:<{jti}>
• Stored Value: A serialized RefreshToken object, including fields like created_at, refresh_token, and user-agent
• Stored Key: auth:permission:<{user_id}>
• Stored Value: Current user's permission code

This structure is extensible — you can include IP address verification, device ID, platform identifier, restrict refresh sources, or implement multi-device login control strategies.

All dependencies and logic are injected via type annotations and FastAPI’s dependency system, making them easy to reuse and extend.

---

Websocket Socket.IO Server Wrapper with Dependency Injection

This module wraps socketio.AsyncServer, introducing a FastAPI-style dependency injection system. It automatically handles lifecycle management, parameter injection, validation error responses, and the injection of special parameters (such as SID and environ).

Compared to traditional event handling, you no longer need to manually extract data from args or manage dependency initialization and cleanup logic, significantly improving development efficiency and code readability.

Features

• FastAPI-style dependency injection (supports Depends()) and is compatible with fastapi.Depends
• Supports asynchronous dependencies and generator dependencies (yield)
• Automatic lifecycle management with teardown registration
• Pydantic-based parameter validation with error messages returned via error events
• Injects special parameters like SID (session identifier) and environ (Socket.IO request context)
• Automatically parses incoming data into Pydantic models

Example

from pydantic import BaseModel
from src.websockets.server import AsyncServer
from src.websockets.params import Depends, SID

sio = AsyncServer()


# Define a Pydantic request model
class ChatMessage(BaseModel):
    text: str
    room: str


# Define a dependency (supports async/generator/sync)
async def get_current_user(sid: SID) -> str:
    return f"user_{sid[-4:]}"  # Simulate fetching user info from SID


# Register event with injected dependencies
@sio.on("chat.send")
async def handle_chat(message: ChatMessage, user: str = Depends(get_current_user)):
    print(f"[{user}] sends message to {message.room}: {message.text}")
    await sio.emit("chat.receive", {"user": user, "text": message.text}, room=message.room)

Automatic Error Handling

The following errors are automatically handled and returned to the client via an error event:

• Parameter validation errors (via Pydantic)
• Type errors (e.g., non-dict input for events)

Error response format:

{
  "code": 1007,
  "event": "chat.send",
  "message": "Data Validation Error.",
  "data": "xx field required"
}

Special Parameter Injection：

Parameter	Description
SID	Current connection SID
Environ	Request context (e.g., headers)

from src.websockets.params import SID, Environ


@sio.event
async def connect(sid: SID, environ: Environ):
    print(f"SID: {sid}, UA: {environ['HTTP_USER_AGENT']}")

environ is only available during the "connect" event.

This behavior aligns with the Socket.IO design: environ contains HTTP handshake data, which is only passed during connect(sid, environ) and cannot be accessed in later events. To use environ in other events, you must cache it manually during connection.

---

Celery

DatabaseScheduler — Dynamic Database-based Scheduler

The custom DatabaseScheduler dynamically loads periodic tasks from the database, refreshing at configurable intervals:

• Works like django-celery-beat but framework-agnostic (built for FastAPI)
• Loads tasks periodically (e.g., every 60 seconds) without restarting workers
• Automatically merges with configuration-defined tasks
• Fully async compatible via AsyncSession and asyncpg

Example

from src.core.config import settings
from src.queues.celery import Celery

REDIS_URL = str(settings.CELERY_REDIS_URL)
DATABASE_URL = "postgresql+asyncpg://your_username:your_password@localhost:27017/your_database"

app = Celery("celery_app", broker=REDIS_URL, backend=REDIS_URL)
app.conf.update({"timezone": settings.CELERY_TIMEZONE, "database_url": DATABASE_URL, "refresh_interval": 60})

app.autodiscover_tasks(["src.queues.tasks"])

Run beat with: celery -A "src.queues.app" beat -S "src.queues.scheduler:AsyncDatabaseScheduler" -l info

AsyncTask — Native Async Task Support

Our custom task base class auto-detects async def functions and handles execution:

• Automatically runs async def tasks in the proper event loop
• No need to manually distinguish between sync and async tasks
• Backward compatible with sync tasks

Example

import asyncio
from src.queues.app import app

@app.task
async def run_async_task() -> None:
    print("async task start.")
    await asyncio.sleep(10)
    print("async task done.")

Run worker with: celery -A "src.queues.app" worker -l info

TypedCelery — IDE-Friendly Celery Wrapper

Enhances native Celery with improved type hinting and IDE integration:

• Refactored class definitions to return accurate types
• Enables smart autocomplete and error detection in IDEs like PyCharm or VSCode
• Greatly improves development speed and reduces bugs in large teams

Example 1

Example 2

Example 3

For more details, please refer to the source code in the src.queues directory, including task registration, scheduler implementation, and async task execution logic.

---

Running Tests (with Pytest)

Before running the tests, please ensure you’ve completed the following setup:

1. Copy Environment Configuration：

   cp .env.pytest.example .env.pytest

2. Configure Database Connections

   You can either manually configure .env.pytest, or simply start the pre-configured test database containers using Docker.

   Option A: Start test databases with Docker

    docker compose -f docker-compose-pytest.yml up -d --build

   Option B: Manually configure .env.pytest

    # MongoDB (required)
    MONGO_DATABASE_URL=mongodb://localhost:27017
   
    # Redis (required)
    REDIS_DATABASE_URL=redis://localhost:6379
   
    # SQLite (default relational database, optional)
    SQL_DATABASE_URL=sqlite+aiosqlite://

3. Run the Tests

    pytest -s

4. Run your tests with coverage

    # Run tests and collect coverage data
    coverage run -m --source=src pytest -s tests/
   
    # Display a brief coverage report
    coverage report
   
    # Generate an HTML coverage report in htmlcov/index.html
    coverage html
   
    # Erase previous coverage data
    coverage erase

---

Git Commit Convention

See Git Guidelines

---

License

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

---