1# main.py
 2from fastapi import FastAPI, Depends
 3from fastapi.middleware.cors import CORSMiddleware
 4from contextlib import asynccontextmanager
 5
 6@asynccontextmanager
 7async def lifespan(app: FastAPI):
 8    """Application lifespan events."""
 9    # Startup
10    await database.connect()
11    yield
12    # Shutdown
13    await database.disconnect()
14
15app = FastAPI(
16    title="API Template",
17    version="1.0.0",
18    lifespan=lifespan
19)
20
21# CORS middleware
22app.add_middleware(
23    CORSMiddleware,
24    allow_origins=["*"],
25    allow_credentials=True,
26    allow_methods=["*"],
27    allow_headers=["*"],
28)
29
30# Include routers
31from app.api.v1.router import api_router
32app.include_router(api_router, prefix="/api/v1")
33
34# core/config.py
35from pydantic_settings import BaseSettings
36from functools import lru_cache
37
38class Settings(BaseSettings):
39    """Application settings."""
40    DATABASE_URL: str
41    SECRET_KEY: str
42    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
43    API_V1_STR: str = "/api/v1"
44
45    class Config:
46        env_file = ".env"
47
48@lru_cache()
49def get_settings() -> Settings:
50    return Settings()
51
52# core/database.py
53from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
54from sqlalchemy.ext.declarative import declarative_base
55from sqlalchemy.orm import sessionmaker
56from app.core.config import get_settings
57
58settings = get_settings()
59
60engine = create_async_engine(
61    settings.DATABASE_URL,
62    echo=True,
63    future=True
64)
65
66AsyncSessionLocal = sessionmaker(
67    engine,
68    class_=AsyncSession,
69    expire_on_commit=False
70)
71
72Base = declarative_base()
73
74async def get_db() -> AsyncSession:
75    """Dependency for database session."""
76    async with AsyncSessionLocal() as session:
77        try:
78            yield session
79            await session.commit()
80        except Exception:
81            await session.rollback()
82            raise
83        finally:
84            await session.close()

 1# repositories/base_repository.py
 2from typing import Generic, TypeVar, Type, Optional, List
 3from sqlalchemy.ext.asyncio import AsyncSession
 4from sqlalchemy import select
 5from pydantic import BaseModel
 6
 7ModelType = TypeVar("ModelType")
 8CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
 9UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
10
11class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
12    """Base repository for CRUD operations."""
13
14    def __init__(self, model: Type[ModelType]):
15        self.model = model
16
17    async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
18        """Get by ID."""
19        result = await db.execute(
20            select(self.model).where(self.model.id == id)
21        )
22        return result.scalars().first()
23
24    async def get_multi(
25        self,
26        db: AsyncSession,
27        skip: int = 0,
28        limit: int = 100
29    ) -> List[ModelType]:
30        """Get multiple records."""
31        result = await db.execute(
32            select(self.model).offset(skip).limit(limit)
33        )
34        return result.scalars().all()
35
36    async def create(
37        self,
38        db: AsyncSession,
39        obj_in: CreateSchemaType
40    ) -> ModelType:
41        """Create new record."""
42        db_obj = self.model(**obj_in.dict())
43        db.add(db_obj)
44        await db.flush()
45        await db.refresh(db_obj)
46        return db_obj
47
48    async def update(
49        self,
50        db: AsyncSession,
51        db_obj: ModelType,
52        obj_in: UpdateSchemaType
53    ) -> ModelType:
54        """Update record."""
55        update_data = obj_in.dict(exclude_unset=True)
56        for field, value in update_data.items():
57            setattr(db_obj, field, value)
58        await db.flush()
59        await db.refresh(db_obj)
60        return db_obj
61
62    async def delete(self, db: AsyncSession, id: int) -> bool:
63        """Delete record."""
64        obj = await self.get(db, id)
65        if obj:
66            await db.delete(obj)
67            return True
68        return False
69
70# repositories/user_repository.py
71from app.repositories.base_repository import BaseRepository
72from app.models.user import User
73from app.schemas.user import UserCreate, UserUpdate
74
75class UserRepository(BaseRepository[User, UserCreate, UserUpdate]):
76    """User-specific repository."""
77
78    async def get_by_email(self, db: AsyncSession, email: str) -> Optional[User]:
79        """Get user by email."""
80        result = await db.execute(
81            select(User).where(User.email == email)
82        )
83        return result.scalars().first()
84
85    async def is_active(self, db: AsyncSession, user_id: int) -> bool:
86        """Check if user is active."""
87        user = await self.get(db, user_id)
88        return user.is_active if user else False
89
90user_repository = UserRepository(User)

 1# services/user_service.py
 2from typing import Optional
 3from sqlalchemy.ext.asyncio import AsyncSession
 4from app.repositories.user_repository import user_repository
 5from app.schemas.user import UserCreate, UserUpdate, User
 6from app.core.security import get_password_hash, verify_password
 7
 8class UserService:
 9    """Business logic for users."""
10
11    def __init__(self):
12        self.repository = user_repository
13
14    async def create_user(
15        self,
16        db: AsyncSession,
17        user_in: UserCreate
18    ) -> User:
19        """Create new user with hashed password."""
20        # Check if email exists
21        existing = await self.repository.get_by_email(db, user_in.email)
22        if existing:
23            raise ValueError("Email already registered")
24
25        # Hash password
26        user_in_dict = user_in.dict()
27        user_in_dict["hashed_password"] = get_password_hash(user_in_dict.pop("password"))
28
29        # Create user
30        user = await self.repository.create(db, UserCreate(**user_in_dict))
31        return user
32
33    async def authenticate(
34        self,
35        db: AsyncSession,
36        email: str,
37        password: str
38    ) -> Optional[User]:
39        """Authenticate user."""
40        user = await self.repository.get_by_email(db, email)
41        if not user:
42            return None
43        if not verify_password(password, user.hashed_password):
44            return None
45        return user
46
47    async def update_user(
48        self,
49        db: AsyncSession,
50        user_id: int,
51        user_in: UserUpdate
52    ) -> Optional[User]:
53        """Update user."""
54        user = await self.repository.get(db, user_id)
55        if not user:
56            return None
57
58        if user_in.password:
59            user_in_dict = user_in.dict(exclude_unset=True)
60            user_in_dict["hashed_password"] = get_password_hash(
61                user_in_dict.pop("password")
62            )
63            user_in = UserUpdate(**user_in_dict)
64
65        return await self.repository.update(db, user, user_in)
66
67user_service = UserService()

 1# api/v1/endpoints/users.py
 2from fastapi import APIRouter, Depends, HTTPException, status
 3from sqlalchemy.ext.asyncio import AsyncSession
 4from typing import List
 5
 6from app.core.database import get_db
 7from app.schemas.user import User, UserCreate, UserUpdate
 8from app.services.user_service import user_service
 9from app.api.dependencies import get_current_user
10
11router = APIRouter()
12
13@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
14async def create_user(
15    user_in: UserCreate,
16    db: AsyncSession = Depends(get_db)
17):
18    """Create new user."""
19    try:
20        user = await user_service.create_user(db, user_in)
21        return user
22    except ValueError as e:
23        raise HTTPException(status_code=400, detail=str(e))
24
25@router.get("/me", response_model=User)
26async def read_current_user(
27    current_user: User = Depends(get_current_user)
28):
29    """Get current user."""
30    return current_user
31
32@router.get("/{user_id}", response_model=User)
33async def read_user(
34    user_id: int,
35    db: AsyncSession = Depends(get_db),
36    current_user: User = Depends(get_current_user)
37):
38    """Get user by ID."""
39    user = await user_service.repository.get(db, user_id)
40    if not user:
41        raise HTTPException(status_code=404, detail="User not found")
42    return user
43
44@router.patch("/{user_id}", response_model=User)
45async def update_user(
46    user_id: int,
47    user_in: UserUpdate,
48    db: AsyncSession = Depends(get_db),
49    current_user: User = Depends(get_current_user)
50):
51    """Update user."""
52    if current_user.id != user_id:
53        raise HTTPException(status_code=403, detail="Not authorized")
54
55    user = await user_service.update_user(db, user_id, user_in)
56    if not user:
57        raise HTTPException(status_code=404, detail="User not found")
58    return user
59
60@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
61async def delete_user(
62    user_id: int,
63    db: AsyncSession = Depends(get_db),
64    current_user: User = Depends(get_current_user)
65):
66    """Delete user."""
67    if current_user.id != user_id:
68        raise HTTPException(status_code=403, detail="Not authorized")
69
70    deleted = await user_service.repository.delete(db, user_id)
71    if not deleted:
72        raise HTTPException(status_code=404, detail="User not found")

 1# core/security.py
 2from datetime import datetime, timedelta
 3from typing import Optional
 4from jose import JWTError, jwt
 5from passlib.context import CryptContext
 6from app.core.config import get_settings
 7
 8settings = get_settings()
 9pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
10
11ALGORITHM = "HS256"
12
13def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
14    """Create JWT access token."""
15    to_encode = data.copy()
16    if expires_delta:
17        expire = datetime.utcnow() + expires_delta
18    else:
19        expire = datetime.utcnow() + timedelta(minutes=15)
20    to_encode.update({"exp": expire})
21    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
22    return encoded_jwt
23
24def verify_password(plain_password: str, hashed_password: str) -> bool:
25    """Verify password against hash."""
26    return pwd_context.verify(plain_password, hashed_password)
27
28def get_password_hash(password: str) -> str:
29    """Hash password."""
30    return pwd_context.hash(password)
31
32# api/dependencies.py
33from fastapi import Depends, HTTPException, status
34from fastapi.security import OAuth2PasswordBearer
35from jose import JWTError, jwt
36from sqlalchemy.ext.asyncio import AsyncSession
37
38from app.core.database import get_db
39from app.core.security import ALGORITHM
40from app.core.config import get_settings
41from app.repositories.user_repository import user_repository
42
43oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
44
45async def get_current_user(
46    db: AsyncSession = Depends(get_db),
47    token: str = Depends(oauth2_scheme)
48):
49    """Get current authenticated user."""
50    credentials_exception = HTTPException(
51        status_code=status.HTTP_401_UNAUTHORIZED,
52        detail="Could not validate credentials",
53        headers={"WWW-Authenticate": "Bearer"},
54    )
55
56    try:
57        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
58        user_id: int = payload.get("sub")
59        if user_id is None:
60            raise credentials_exception
61    except JWTError:
62        raise credentials_exception
63
64    user = await user_repository.get(db, user_id)
65    if user is None:
66        raise credentials_exception
67
68    return user

 1# tests/conftest.py
 2import pytest
 3import asyncio
 4from httpx import AsyncClient
 5from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
 6from sqlalchemy.orm import sessionmaker
 7
 8from app.main import app
 9from app.core.database import get_db, Base
10
11TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
12
13@pytest.fixture(scope="session")
14def event_loop():
15    loop = asyncio.get_event_loop_policy().new_event_loop()
16    yield loop
17    loop.close()
18
19@pytest.fixture
20async def db_session():
21    engine = create_async_engine(TEST_DATABASE_URL, echo=True)
22    async with engine.begin() as conn:
23        await conn.run_sync(Base.metadata.create_all)
24
25    AsyncSessionLocal = sessionmaker(
26        engine, class_=AsyncSession, expire_on_commit=False
27    )
28
29    async with AsyncSessionLocal() as session:
30        yield session
31
32@pytest.fixture
33async def client(db_session):
34    async def override_get_db():
35        yield db_session
36
37    app.dependency_overrides[get_db] = override_get_db
38
39    async with AsyncClient(app=app, base_url="http://test") as client:
40        yield client
41
42# tests/test_users.py
43import pytest
44
45@pytest.mark.asyncio
46async def test_create_user(client):
47    response = await client.post(
48        "/api/v1/users/",
49        json={
50            "email": "test@example.com",
51            "password": "testpass123",
52            "name": "Test User"
53        }
54    )
55    assert response.status_code == 201
56    data = response.json()
57    assert data["email"] == "test@example.com"
58    assert "id" in data