বড় Project এর Structure

notes-api/
├── pyproject.toml
├── alembic.ini
├── .env.example
├── migrations/
├── src/app/
│   ├── main.py                 ← FastAPI app তৈরি, router যোগ, middleware
│   ├── core/
│   │   ├── config.py           ← Settings (pydantic-settings)
│   │   ├── db.py               ← engine, session
│   │   ├── security.py         ← password hash, JWT
│   │   └── errors.py           ← AppError পরিবার + exception handler
│   ├── models/                 ← SQLAlchemy model
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── note.py
│   ├── schemas/                ← Pydantic (API এর input/output)
│   │   ├── user.py
│   │   └── note.py
│   ├── repositories/
│   │   └── note_repo.py
│   ├── services/
│   │   └── note_service.py
│   ├── api/
│   │   ├── deps.py             ← Depends: db session, current user
│   │   └── v1/
│   │       ├── __init__.py     ← v1 এর সব router এক জায়গায়
│   │       ├── auth.py
│   │       └── notes.py
│   └── workers/                ← background job (পরে)
└── tests/
    ├── conftest.py
    ├── test_notes_api.py
    └── test_note_service.py

# core/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    app_name: str = "Notes API"
    environment: str = "development"          # development / staging / production
    database_url: str
    jwt_secret: str
    jwt_access_minutes: int = 15
    jwt_refresh_days: int = 30
    cors_origins: list[str] = []


@lru_cache
def get_settings() -> Settings:
    return Settings()

# core/db.py
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.core.config import get_settings

engine = create_async_engine(get_settings().database_url, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def get_db() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session            # request শেষে session বন্ধ, error হলেও

# api/deps.py
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.db import get_db

DbDep = Annotated[AsyncSession, Depends(get_db)]

# repositories/note_repo.py
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.note import Note


class NoteRepository:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def get(self, note_id: int) -> Note | None:
        return await self.db.get(Note, note_id)

    async def list_for_user(self, user_id: int, limit: int, offset: int) -> list[Note]:
        stmt = (
            select(Note)
            .where(Note.owner_id == user_id, Note.deleted_at.is_(None))
            .order_by(Note.updated_at.desc())
            .limit(limit)
            .offset(offset)
        )
        return list(await self.db.scalars(stmt))

    async def count_for_user(self, user_id: int) -> int:
        stmt = select(func.count()).select_from(Note).where(Note.owner_id == user_id, Note.deleted_at.is_(None))
        return await self.db.scalar(stmt) or 0

    def add(self, note: Note) -> None:
        self.db.add(note)

# services/note_service.py
from datetime import datetime, timezone
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.errors import NotFoundError, PermissionDeniedError, LimitExceededError
from app.models.note import Note
from app.repositories.note_repo import NoteRepository
from app.schemas.note import NoteCreate, NoteUpdate

MAX_NOTES_FREE = 500


class NoteService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.notes = NoteRepository(db)

    async def create(self, user_id: int, data: NoteCreate) -> Note:
        if await self.notes.count_for_user(user_id) >= MAX_NOTES_FREE:
            raise LimitExceededError("Free plan এ সর্বোচ্চ ৫০০টা note")
        note = Note(owner_id=user_id, **data.model_dump())
        self.notes.add(note)
        await self.db.commit()
        await self.db.refresh(note)
        return note

    async def get_owned(self, user_id: int, note_id: int) -> Note:
        note = await self.notes.get(note_id)
        if note is None or note.deleted_at is not None:
            raise NotFoundError("note", note_id)
        if note.owner_id != user_id:
            raise PermissionDeniedError()          # অন্যের note
        return note

    async def update(self, user_id: int, note_id: int, data: NoteUpdate) -> Note:
        note = await self.get_owned(user_id, note_id)
        for field, value in data.model_dump(exclude_unset=True).items():
            setattr(note, field, value)
        await self.db.commit()
        await self.db.refresh(note)
        return note

    async def delete(self, user_id: int, note_id: int) -> None:
        note = await self.get_owned(user_id, note_id)
        note.deleted_at = datetime.now(timezone.utc)      # soft delete
        await self.db.commit()

# core/errors.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class AppError(Exception):
    status_code = 400
    code = "APP_ERROR"

    def __init__(self, message: str = "") -> None:
        self.message = message or self.__class__.__doc__ or "সমস্যা হয়েছে"
        super().__init__(self.message)


class NotFoundError(AppError):
    """পাওয়া যায়নি"""
    status_code = 404
    code = "NOT_FOUND"

    def __init__(self, resource: str, resource_id: int) -> None:
        super().__init__(f"{resource} #{resource_id} পাওয়া যায়নি")


class PermissionDeniedError(AppError):
    """এই কাজের অনুমতি নেই"""
    status_code = 403
    code = "FORBIDDEN"


class LimitExceededError(AppError):
    status_code = 422
    code = "LIMIT_EXCEEDED"


def register_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def handle_app_error(request: Request, exc: AppError):
        return JSONResponse(
            status_code=exc.status_code,
            content={"error": {"code": exc.code, "message": exc.message}},
        )

# api/v1/notes.py
from typing import Annotated
from fastapi import APIRouter, Depends, status
from app.api.deps import DbDep, CurrentUser, PageDep
from app.schemas.note import NoteCreate, NoteUpdate, NoteOut, NotePage
from app.services.note_service import NoteService

router = APIRouter(prefix="/notes", tags=["notes"])


def get_service(db: DbDep) -> NoteService:
    return NoteService(db)

ServiceDep = Annotated[NoteService, Depends(get_service)]


@router.post("", response_model=NoteOut, status_code=status.HTTP_201_CREATED)
async def create_note(data: NoteCreate, user: CurrentUser, svc: ServiceDep):
    return await svc.create(user.id, data)


@router.get("/{note_id}", response_model=NoteOut)
async def get_note(note_id: int, user: CurrentUser, svc: ServiceDep):
    return await svc.get_owned(user.id, note_id)


@router.patch("/{note_id}", response_model=NoteOut)
async def update_note(note_id: int, data: NoteUpdate, user: CurrentUser, svc: ServiceDep):
    return await svc.update(user.id, note_id, data)


@router.delete("/{note_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_note(note_id: int, user: CurrentUser, svc: ServiceDep):
    await svc.delete(user.id, note_id)

# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1 import api_router
from app.core.config import get_settings
from app.core.errors import register_error_handlers


def create_app() -> FastAPI:
    settings = get_settings()
    app = FastAPI(
        title=settings.app_name,
        docs_url=None if settings.environment == "production" else "/docs",
    )
    app.add_middleware(CORSMiddleware, allow_origins=settings.cors_origins, allow_methods=["*"], allow_headers=["*"])
    register_error_handlers(app)
    app.include_router(api_router, prefix="/v1")

    @app.get("/health/live", include_in_schema=False)
    async def live():
        return {"status": "ok"}

    return app


app = create_app()