wehub-resource-sync
18 KiB
18 KiB
FastAPI Database Integration
Complete database setup with SQLAlchemy, Alembic, and async support for FastAPI.
Usage
# Initialize Alembic
alembic init alembic
# Create migration
alembic revision --autogenerate -m "Initial migration"
# Apply migrations
alembic upgrade head
# Downgrade migration
alembic downgrade -1
Database Configuration
# app/core/config.py
from pydantic import BaseSettings, PostgresDsn, validator
from typing import Optional, Dict, Any
import os
class Settings(BaseSettings):
"""Application settings."""
# Database
POSTGRES_SERVER: str = "localhost"
POSTGRES_USER: str = "postgres"
POSTGRES_PASSWORD: str = "password"
POSTGRES_DB: str = "fastapi_app"
POSTGRES_PORT: str = "5432"
DATABASE_URL: Optional[PostgresDsn] = None
@validator("DATABASE_URL", pre=True)
def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any:
if isinstance(v, str):
return v
return PostgresDsn.build(
scheme="postgresql+asyncpg",
user=values.get("POSTGRES_USER"),
password=values.get("POSTGRES_PASSWORD"),
host=values.get("POSTGRES_SERVER"),
port=values.get("POSTGRES_PORT"),
path=f"/{values.get('POSTGRES_DB') or ''}",
)
# Redis
REDIS_URL: str = "redis://localhost:6379/0"
# Database settings
DATABASE_POOL_SIZE: int = 10
DATABASE_MAX_OVERFLOW: int = 20
DATABASE_POOL_RECYCLE: int = 3600
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()
Database Setup
# app/db/database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
# Create async engine
engine = create_async_engine(
str(settings.DATABASE_URL),
pool_size=settings.DATABASE_POOL_SIZE,
max_overflow=settings.DATABASE_MAX_OVERFLOW,
pool_recycle=settings.DATABASE_POOL_RECYCLE,
pool_pre_ping=True,
echo=False # Set to True for SQL debugging
)
# Create async session factory
AsyncSessionLocal = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
# Base class for models
Base = declarative_base()
async def get_db() -> AsyncSession:
"""Dependency to get database session."""
async with AsyncSessionLocal() as session:
try:
yield session
finally:
await session.close()
async def create_tables():
"""Create database tables."""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async def drop_tables():
"""Drop database tables."""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
Base Model
# app/models/base.py
from sqlalchemy import Column, Integer, DateTime, func
from sqlalchemy.ext.declarative import declared_attr
from app.db.database import Base
from datetime import datetime
from typing import Any
class TimestampMixin:
"""Mixin for timestamp fields."""
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False
)
class BaseModel(Base, TimestampMixin):
"""Base model with common functionality."""
__abstract__ = True
id = Column(Integer, primary_key=True, index=True)
@declared_attr
def __tablename__(cls) -> str:
return cls.__name__.lower()
def dict(self, exclude: set = None) -> dict[str, Any]:
"""Convert model to dictionary."""
exclude = exclude or set()
return {
column.name: getattr(self, column.name)
for column in self.__table__.columns
if column.name not in exclude
}
def __repr__(self) -> str:
return f"<{self.__class__.__name__}(id={self.id})>"
Example Models
# app/models/user.py
from sqlalchemy import Column, String, Boolean, Text, Index
from sqlalchemy.orm import relationship
from app.models.base import BaseModel
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(BaseModel):
"""User model."""
__tablename__ = "users"
username = Column(String(50), unique=True, index=True, nullable=False)
email = Column(String(100), unique=True, index=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
first_name = Column(String(50), nullable=False)
last_name = Column(String(50), nullable=False)
is_active = Column(Boolean, default=True, nullable=False)
is_superuser = Column(Boolean, default=False, nullable=False)
bio = Column(Text)
# Relationships
posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")
# Indexes
__table_args__ = (
Index('idx_user_email_active', email, is_active),
Index('idx_user_username_active', username, is_active),
)
def verify_password(self, password: str) -> bool:
"""Verify password against hash."""
return pwd_context.verify(password, self.hashed_password)
@staticmethod
def get_password_hash(password: str) -> str:
"""Generate password hash."""
return pwd_context.hash(password)
def set_password(self, password: str) -> None:
"""Set user password."""
self.hashed_password = self.get_password_hash(password)
@property
def full_name(self) -> str:
"""Get user's full name."""
return f"{self.first_name} {self.last_name}"
def dict(self, exclude: set = None) -> dict:
"""Convert to dict excluding sensitive data."""
exclude = exclude or set()
exclude.add('hashed_password')
return super().dict(exclude=exclude)
# app/models/post.py
from sqlalchemy import Column, String, Text, ForeignKey, Boolean, Index
from sqlalchemy.orm import relationship
from app.models.base import BaseModel
class Post(BaseModel):
"""Blog post model."""
__tablename__ = "posts"
title = Column(String(200), nullable=False, index=True)
content = Column(Text, nullable=False)
slug = Column(String(200), unique=True, nullable=False, index=True)
is_published = Column(Boolean, default=False, nullable=False)
# Foreign keys
author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
# Relationships
author = relationship("User", back_populates="posts")
# Indexes
__table_args__ = (
Index('idx_post_published_created', is_published, 'created_at'),
Index('idx_post_author_published', author_id, is_published),
)
Repository Pattern
# app/repositories/base.py
from typing import Generic, TypeVar, Type, Optional, List, Dict, Any
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, delete, func
from sqlalchemy.orm import selectinload
from app.models.base import BaseModel
ModelType = TypeVar("ModelType", bound=BaseModel)
class BaseRepository(Generic[ModelType]):
"""Base repository with common CRUD operations."""
def __init__(self, model: Type[ModelType], db: AsyncSession):
self.model = model
self.db = db
async def get(self, id: int) -> Optional[ModelType]:
"""Get model by ID."""
result = await self.db.execute(
select(self.model).where(self.model.id == id)
)
return result.scalar_one_or_none()
async def get_multi(
self,
skip: int = 0,
limit: int = 100,
filters: Dict[str, Any] = None
) -> List[ModelType]:
"""Get multiple models with pagination."""
query = select(self.model)
if filters:
for field, value in filters.items():
if hasattr(self.model, field):
query = query.where(getattr(self.model, field) == value)
query = query.offset(skip).limit(limit)
result = await self.db.execute(query)
return result.scalars().all()
async def create(self, obj_in: Dict[str, Any]) -> ModelType:
"""Create new model."""
db_obj = self.model(**obj_in)
self.db.add(db_obj)
await self.db.commit()
await self.db.refresh(db_obj)
return db_obj
async def update(
self,
id: int,
obj_in: Dict[str, Any]
) -> Optional[ModelType]:
"""Update model by ID."""
await self.db.execute(
update(self.model)
.where(self.model.id == id)
.values(**obj_in)
)
await self.db.commit()
return await self.get(id)
async def delete(self, id: int) -> bool:
"""Delete model by ID."""
result = await self.db.execute(
delete(self.model).where(self.model.id == id)
)
await self.db.commit()
return result.rowcount > 0
async def count(self, filters: Dict[str, Any] = None) -> int:
"""Count models with optional filters."""
query = select(func.count(self.model.id))
if filters:
for field, value in filters.items():
if hasattr(self.model, field):
query = query.where(getattr(self.model, field) == value)
result = await self.db.execute(query)
return result.scalar()
# app/repositories/user.py
from typing import Optional
from sqlalchemy import select
from app.models.user import User
from app.repositories.base import BaseRepository
class UserRepository(BaseRepository[User]):
"""User repository with custom methods."""
async def get_by_email(self, email: str) -> Optional[User]:
"""Get user by email."""
result = await self.db.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()
async def get_by_username(self, username: str) -> Optional[User]:
"""Get user by username."""
result = await self.db.execute(
select(User).where(User.username == username)
)
return result.scalar_one_or_none()
async def get_active_users(self, skip: int = 0, limit: int = 100):
"""Get active users."""
return await self.get_multi(
skip=skip,
limit=limit,
filters={'is_active': True}
)
Alembic Configuration
# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from sqlalchemy.ext.asyncio import AsyncEngine
from alembic import context
import asyncio
# Import your models
from app.models.base import Base
from app.models.user import User
from app.models.post import Post
from app.core.config import settings
# Alembic Config object
config = context.config
# Override database URL
config.set_main_option("sqlalchemy.url", str(settings.DATABASE_URL))
# Interpret the config file for logging
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# Add your model's MetaData object here
target_metadata = Base.metadata
def do_run_migrations(connection):
context.configure(
connection=connection,
target_metadata=target_metadata,
compare_type=True,
compare_server_default=True,
)
with context.begin_transaction():
context.run_migrations()
async def run_migrations_online():
"""Run migrations in 'online' mode."""
configuration = config.get_section(config.config_ini_section)
configuration["sqlalchemy.url"] = str(settings.DATABASE_URL)
connectable = AsyncEngine(
engine_from_config(
configuration,
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
if context.is_offline_mode():
run_migrations_offline()
else:
asyncio.run(run_migrations_online())
Database Utilities
# app/db/utils.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
from app.db.database import engine, AsyncSessionLocal
from app.core.config import settings
import asyncio
async def check_database_connection() -> bool:
"""Check if database is accessible."""
try:
async with AsyncSessionLocal() as session:
await session.execute(text("SELECT 1"))
return True
except Exception:
return False
async def create_database_if_not_exists():
"""Create database if it doesn't exist."""
# This is PostgreSQL specific
import asyncpg
from urllib.parse import urlparse
url = urlparse(str(settings.DATABASE_URL))
try:
# Connect to postgres database to create our database
conn = await asyncpg.connect(
host=url.hostname,
port=url.port,
user=url.username,
password=url.password,
database='postgres'
)
# Check if database exists
exists = await conn.fetchval(
"SELECT 1 FROM pg_database WHERE datname = $1",
url.path[1:] # Remove leading slash
)
if not exists:
await conn.execute(f'CREATE DATABASE "{url.path[1:]}"')
print(f"Database {url.path[1:]} created.")
await conn.close()
except Exception as e:
print(f"Error creating database: {e}")
async def execute_raw_sql(sql: str, params: dict = None) -> list:
"""Execute raw SQL query."""
async with AsyncSessionLocal() as session:
result = await session.execute(text(sql), params or {})
return result.fetchall()
async def get_table_info(table_name: str) -> dict:
"""Get information about a table."""
sql = """
SELECT
column_name,
data_type,
is_nullable,
column_default
FROM information_schema.columns
WHERE table_name = :table_name
ORDER BY ordinal_position;
"""
result = await execute_raw_sql(sql, {'table_name': table_name})
return [
{
'column_name': row[0],
'data_type': row[1],
'is_nullable': row[2],
'column_default': row[3]
}
for row in result
]
Database Initialization
# app/db/init_db.py
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.database import get_db, create_tables
from app.models.user import User
from app.repositories.user import UserRepository
from app.core.config import settings
import asyncio
async def init_db() -> None:
"""Initialize database with tables and default data."""
# Create tables
await create_tables()
print("Database tables created.")
# Create default superuser
async with AsyncSessionLocal() as session:
user_repo = UserRepository(User, session)
# Check if superuser exists
existing_user = await user_repo.get_by_email("admin@example.com")
if not existing_user:
superuser_data = {
"username": "admin",
"email": "admin@example.com",
"first_name": "Admin",
"last_name": "User",
"is_superuser": True,
"is_active": True
}
superuser = User(**superuser_data)
superuser.set_password("admin123")
session.add(superuser)
await session.commit()
print("Superuser created.")
else:
print("Superuser already exists.")
if __name__ == "__main__":
asyncio.run(init_db())
Testing Database
# tests/conftest.py
import pytest
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.db.database import Base, get_db
from app.main import app
import pytest_asyncio
# Test database URL
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
@pytest.fixture(scope="session")
def event_loop():
"""Create event loop for async tests."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(scope="session")
async def test_engine():
"""Create test database engine."""
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
# Create tables
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
# Drop tables
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await engine.dispose()
@pytest_asyncio.fixture
async def test_session(test_engine):
"""Create test database session."""
TestSessionLocal = sessionmaker(
test_engine, class_=AsyncSession, expire_on_commit=False
)
async with TestSessionLocal() as session:
yield session
@pytest.fixture
def override_get_db(test_session):
"""Override database dependency."""
async def _override_get_db():
yield test_session
app.dependency_overrides[get_db] = _override_get_db
yield
app.dependency_overrides = {}
Database Health Check
# app/api/health.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
from app.db.database import get_db
import time
router = APIRouter()
@router.get("/health")
async def health_check(db: AsyncSession = Depends(get_db)):
"""Health check endpoint."""
checks = {
"status": "healthy",
"timestamp": time.time(),
"database": await check_database_health(db),
}
# Determine overall status
if checks["database"]["status"] != "ok":
checks["status"] = "unhealthy"
raise HTTPException(status_code=503, detail=checks)
return checks
async def check_database_health(db: AsyncSession) -> dict:
"""Check database connection."""
try:
start_time = time.time()
await db.execute(text("SELECT 1"))
response_time = (time.time() - start_time) * 1000 # milliseconds
return {
"status": "ok",
"response_time_ms": round(response_time, 2)
}
except Exception as e:
return {
"status": "error",
"error": str(e)
}