An extendable async API using FastAPI, Pydantic V2, SQLAlchemy 2.0 and PostgreSQL.
from typing import Annotated

from app.core.security import SECRET_KEY, ALGORITHM, oauth2_scheme

from sqlalchemy.ext.asyncio import AsyncSession
from jose import JWTError, jwt
from fastapi import Depends, HTTPException

from app.core.database import async_get_db
from app.core.models import TokenData
from app.models.user import User
from app.crud.crud_users import crud_users
from app.api.exceptions import credentials_exception, privileges_exception

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: Annotated[AsyncSession, Depends(async_get_db)]) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username_or_email: str = payload.get("sub")
        if username_or_email is None:
            raise credentials_exception
        token_data = TokenData(username_or_email=username_or_email)
    except JWTError:
        raise credentials_exception
    
    if "@" in username_or_email:
        user = await crud_users.get(db=db, email=token_data.username_or_email)
    else: 
        user = await crud_users.get(db=db, username=token_data.username_or_email)
    
    if user is None:
        raise credentials_exception
    
    if user.is_deleted:
        raise HTTPException(status_code=400, detail="User deleted")

    return user

def get_current_superuser(current_user: Annotated[User, Depends(get_current_user)]) -> User:
    if not current_user.is_superuser:
        raise privileges_exception
    
    return current_user

src/app/api/dependencies.py
igormagalhaesr/FastAPI-boilerplate
from fastapi import HTTPException, status

credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Incorrect email, username or password",
    headers={"WWW-Authenticate": "Bearer"},
)

privileges_exception = HTTPException(
    status_code=403, 
    detail="The user doesn't have enough privileges"
)

src/app/api/exceptions.py
igormagalhaesr/FastAPI-boilerplate
from fastapi import APIRouter

from app.api.v1.login import router as login_router
from app.api.v1.users import router as users_router
from app.api.v1.posts import router as posts_router

router = APIRouter(prefix="/v1")
router.include_router(login_router)
router.include_router(users_router)
router.include_router(posts_router)

src/app/api/v1/__init__.py
igormagalhaesr/FastAPI-boilerplate
from typing import Annotated
from datetime import timedelta

from fastapi import Depends
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
import fastapi

from app.core.database import async_get_db
from app.core.models import Token
from app.core.security import ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token, authenticate_user
from app.api.exceptions import credentials_exception

router = fastapi.APIRouter(tags=["login"])

@router.post("/login", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: AsyncSession = Depends(async_get_db)
):
    
    user = await authenticate_user(
        username_or_email=form_data.username, 
        password=form_data.password, 
        db=db
    )
    if not user:
        raise credentials_exception
    
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = await create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    
    return {"access_token": access_token, "token_type": "bearer"}

src/app/api/v1/login.py
igormagalhaesr/FastAPI-boilerplate
from typing import List, Annotated

from fastapi import Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
import fastapi

from app.schemas.post import PostCreate, PostUpdate, PostRead, PostCreateInternal
from app.schemas.user import UserRead
from app.api.dependencies import get_current_user, get_current_superuser
from app.core.database import async_get_db
from app.crud.crud_posts import crud_posts
from app.crud.crud_users import crud_users
from app.api.exceptions import privileges_exception

router = fastapi.APIRouter(tags=["posts"])

@router.post("/{username}/post", response_model=PostRead, status_code=201)
async def write_post(
    username: str,
    post: PostCreate, 
    current_user: Annotated[UserRead, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(async_get_db)]
):
    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    if current_user.id != db_user.id:
        raise privileges_exception

    post_internal_dict = post.model_dump()
    post_internal_dict["created_by_user_id"] = db_user.id
    post_internal = PostCreateInternal(**post_internal_dict)
    return await crud_posts.create(db=db, object=post_internal)


@router.get("/{username}/posts", response_model=List[PostRead])
async def read_posts(
    username: str, 
    db: Annotated[AsyncSession, Depends(async_get_db)]
):
    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    posts = await crud_posts.get_multi(db=db, created_by_user_id=db_user.id, is_deleted=False)
    return posts


@router.get("/{username}/post/{id}", response_model=PostRead)
async def read_post(
    username: str,
    id: int, 
    db: Annotated[AsyncSession, Depends(async_get_db)]
):
    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_post = await crud_posts.get(db=db, id=id, created_by_user_id=db_user.id, is_deleted=False)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")

    return db_post


@router.patch("/{username}/post/{id}", response_model=PostRead)
async def patch_post(
    username: str,
    id: int,
    values: PostUpdate,
    current_user: Annotated[UserRead, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(async_get_db)]
):
    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    if current_user.id != db_user.id:
        raise privileges_exception

    db_post = await crud_posts.get(db=db, id=id, is_deleted=False)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    
    return await crud_posts.update(db=db, object=values, db_object=db_post, id=id)


@router.delete("/{username}/post/{id}")
async def erase_post(
    username: str,
    id: int,
    current_user: Annotated[UserRead, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(async_get_db)]
):
    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    if current_user.id != db_user.id:
        raise privileges_exception

    db_post = await crud_posts.get(db=db, id=id, is_deleted=False)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    
    deleted_post = await crud_posts.delete(db=db, db_object=db_post, id=id)
    if deleted_post.is_deleted == True:
        message = {"message": "Post deleted"}
    else:
        message = {"message": "Something went wrong"}

    return message


@router.delete("/{username}/db_post/{id}", dependencies=[Depends(get_current_superuser)])
async def erase_db_post(
    username: str,
    id: int,
    db: Annotated[AsyncSession, Depends(async_get_db)]
):
    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_post = await crud_posts.get(db=db, id=id, is_deleted=False)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    
    await crud_posts.db_delete(db=db, db_object=db_post, id=id)
    deleted_post = await crud_posts.get(db=db, id=id)
    
    if deleted_post is None:
        message = {"message": "Post deleted"}
    else:
        message = {"message": "Something went wrong"}

    return message

src/app/api/v1/posts.py
igormagalhaesr/FastAPI-boilerplate
from typing import List, Annotated

from fastapi import Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
import fastapi

from app.schemas.user import UserCreate, UserCreateInternal, UserUpdate, UserRead, UserBase 
from app.api.dependencies import get_current_user, get_current_superuser
from app.core.database import async_get_db
from app.core.security import get_password_hash
from app.crud.crud_users import crud_users
from app.api.exceptions import privileges_exception

router = fastapi.APIRouter(tags=["users"])

@router.post("/user", response_model=UserRead, status_code=201)
async def write_user(user: UserCreate, db: AsyncSession = Depends(async_get_db)):
    db_user = await crud_users.get(db=db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email is already registered")

    db_user = await crud_users.get(db=db, username=user.username)
    if db_user:
        raise HTTPException(status_code=400, detail="Username not available")
    
    user_internal_dict = user.model_dump()
    user_internal_dict["hashed_password"] = get_password_hash(password=user_internal_dict["password"])
    del user_internal_dict["password"]

    user_internal = UserCreateInternal(**user_internal_dict)
    return await crud_users.create(db=db, object=user_internal)


@router.get("/users", response_model=List[UserRead])
async def read_users(db: AsyncSession = Depends(async_get_db)):
    users = await crud_users.get_multi(db=db, is_deleted=False)
    return users


@router.get("/user/me/", response_model=UserRead)
async def read_users_me(
    current_user: Annotated[UserRead, Depends(get_current_user)]
):
    return current_user


@router.get("/user/{username}", response_model=UserRead)
async def read_user(username: str, db: AsyncSession = Depends(async_get_db)):
    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    return db_user


@router.patch("/user/{username}", response_model=UserRead)
async def patch_user(
    values: UserUpdate,
    username: str,
    current_user: Annotated[UserRead, Depends(get_current_user)],
    db: AsyncSession = Depends(async_get_db)
):
    db_user = await crud_users.get(db=db, username=username)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    if db_user.username != current_user.username:
        raise privileges_exception
    
    if values.username != db_user.username:
        existing_username = await crud_users.get(db=db, username=values.username)
        if existing_username is not None:
            raise HTTPException(status_code=400, detail="Username not available")

    if values.email != db_user.email:
        existing_email = await crud_users.get(db=db, email=values.email)
        if existing_email:
            raise HTTPException(status_code=400, detail="Email is already registered")

    db_user = await crud_users.update(db=db, object=values, db_object=db_user)
    return db_user


@router.delete("/user/{username}")
async def erase_user(
    username: str, 
    current_user: Annotated[UserRead, Depends(get_current_user)],
    db: AsyncSession = Depends(async_get_db)
):
    db_user = await crud_users.get(db=db, username=username)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    if db_user.username != current_user.username:
        raise privileges_exception

    db_user = await crud_users.delete(db=db, db_object=db_user, username=username)
    return db_user


@router.delete("/db_user/{username}", dependencies=[Depends(get_current_superuser)])
async def erase_db_user(
    username: str,
    db: AsyncSession = Depends(async_get_db)
):
    db_user = await crud_users.get(db=db, username=username)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_user = await crud_users.db_delete(db=db, db_object=db_user, username=username)
    return db_user

src/app/api/v1/users.py
igormagalhaesr/FastAPI-boilerplate
from typing import Optional

from decouple import config
from pydantic import SecretStr
from pydantic_settings import BaseSettings

class AppSettings(BaseSettings):
    APP_NAME: str = config("APP_NAME")
    APP_DESCRIPTION: str = config("APP_DESCRIPTION")
    APP_VERSION: str = config("APP_VERSION")
    CONTACT: dict = {
        "name": config("CONTACT_NAME"),
        "email": config("CONTACT_EMAIL")
    }
    LICENSE_NAME: dict = {"name": config("LICENSE_NAME")}


class CryptSettings(BaseSettings):
    SECRET_KEY: str = config("SECRET_KEY")
    ALGORITHM: str = config("ALGORITHM")
    ACCESS_TOKEN_EXPIRE_MINUTES: int = config("ACCESS_TOKEN_EXPIRE_MINUTES")


class PostgresSettings(BaseSettings):
    POSTGRES_USER: str = config("POSTGRES_USER", default="postgres")
    POSTGRES_PASSWORD: str = config("POSTGRES_PASSWORD", default="postgres")
    POSTGRES_SERVER: str = config("POSTGRES_SERVER", default="localhost")
    POSTGRES_PORT: str = config("POSTGRES_PORT", default=5432)
    POSTGRES_DB: str = config("POSTGRES_DB", default="postgres")
    POSTGRES_URI: str = f"{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_SERVER}:{POSTGRES_PORT}/{POSTGRES_DB}"


class FirstUserSettings(BaseSettings):
    ADMIN_NAME: str = config("ADMIN_NAME", default="admin")
    ADMIN_EMAIL: str = config("ADMIN_EMAIL", default="admin@admin.com")
    ADMIN_USERNAME: str = config("ADMIN_USERNAME", default="admin")
    ADMIN_PASSWORD: str = config("ADMIN_PASSWORD", default="!Ch4ng3Th1sP4ssW0rd!")


class Settings(
    AppSettings, 
    PostgresSettings, 
    CryptSettings, 
    FirstUserSettings
):
    pass


settings = Settings()

src/app/core/config.py
igormagalhaesr/FastAPI-boilerplate
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio.session import AsyncSession
from sqlalchemy.orm import DeclarativeBase, sessionmaker, MappedAsDataclass


from app.core.config import settings

class Base(DeclarativeBase, MappedAsDataclass):
    pass

DATABASE_URI = settings.POSTGRES_URI
ASYNC_SQLALCHEMY_DATABASE_URL = f"postgresql+asyncpg://{DATABASE_URI}"

async_engine = create_async_engine(
    ASYNC_SQLALCHEMY_DATABASE_URL, 
    echo=False, 
    future=True
)

local_session = sessionmaker(
    bind=async_engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)

async def async_get_db() -> AsyncSession:
    async_session = local_session
    
    async with async_session() as db:
        yield db
        await db.commit()

src/app/core/database.py
igormagalhaesr/FastAPI-boilerplate
import uuid as uuid_pkg
from datetime import datetime

from pydantic import BaseModel, Field
from sqlalchemy import text

from app.core.database import Base

class HealthCheck(BaseModel):
    name: str
    version: str
    description: str


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username_or_email: str | None = None


class UUIDModel(BaseModel):
    uuid: uuid_pkg.UUID = Field(
        default_factory=uuid_pkg.uuid4,
        json_schema_extra={
            "server_default": text("gen_random_uuid()"),
            "unique": True
        }
    )


class TimestampModel(BaseModel):
    created_at: datetime = Field(
       default_factory=datetime.utcnow,
       json_schema_extra={
           "server_default": text("current_timestamp(0)")
       }
    )

    updated_at: datetime = Field(
       default=None,
       json_schema_extra={
           "server_default": text("current_timestamp(0)"),
           "onupdate": text("current_timestamp(0)")
       }
    )

    
class PersistentDeletion(BaseModel):
    deleted_at: datetime | None = Field(
        default=None,
        json_schema_extra={
            "server_default": text("None"),
            "ondelete": text("current_timestamp(0)")
        }
    )

    is_deleted: bool = False

src/app/core/models.py
igormagalhaesr/FastAPI-boilerplate
from datetime import datetime, timedelta

from sqlalchemy.ext.asyncio import AsyncSession
from passlib.context import CryptContext
from jose import jwt
from fastapi.security import OAuth2PasswordBearer

from app.crud.crud_users import crud_users
from app.core.config import settings


SECRET_KEY = settings.SECRET_KEY
ALGORITHM = settings.ALGORITHM
ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/login")
crypt_context = CryptContext(schemes=["sha256_crypt"])

async def verify_password(plain_password, hashed_password):
    return crypt_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return crypt_context.hash(password)

async def authenticate_user(username_or_email: str, password: str, db: AsyncSession):
    if "@" in username_or_email:
        db_user = await crud_users.get(db=db, email=username_or_email)
    else: 
        db_user = await crud_users.get(db=db, username=username_or_email)
    
    if (not db_user) or (db_user.is_deleted):
        db_user = False
    
    elif not await verify_password(password, db_user.hashed_password):
        db_user = False
    
    return db_user

async def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

src/app/core/security.py
igormagalhaesr/FastAPI-boilerplate
from typing import Any, Dict, Generic, List, Type, TypeVar, Union
from datetime import datetime

from pydantic import BaseModel
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

ModelType = TypeVar("ModelType")
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
UpdateSchemaInternalType = TypeVar("UpdateSchemaInternalType", bound=BaseModel)
DeleteSchemaType = TypeVar("DeleteSchemaType", bound=BaseModel)

class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType, UpdateSchemaInternalType, DeleteSchemaType]):
    def __init__(self, model: Type[ModelType]) -> None:
        self._model = model

    async def create(
            self, db: AsyncSession, object: CreateSchemaType
    ) -> ModelType:
        object_dict = object.model_dump()
        db_object = self._model(**object_dict)
        db.add(db_object)
        await db.commit()
        return db_object

    async def get(self, db: AsyncSession, **kwargs) -> ModelType | None:
        query = select(self._model).filter_by(**kwargs)
        result = await db.execute(query)
        return result.scalar_one_or_none()
    
    async def get_multi(
            self, db: AsyncSession, offset: int = 0, limit: int = 100, **kwargs
    ) -> List[ModelType]:
        query = select(self._model) \
            .filter_by(**kwargs) \
            .offset(offset) \
            .limit(limit)
        
        result = await db.execute(query)
        return result.scalars().all()

    async def update(
            self,
            db: AsyncSession,
            object: Union[UpdateSchemaType, Dict[str, Any]],
            db_object: ModelType | None = None,
            **kwargs
    ) -> ModelType | None:
        db_object = db_object or await self.get(db=db, **kwargs)
        if db_object:
            if isinstance(object, dict):
                update_data = object
            else:
                update_data = object.model_dump(exclude_unset=True)
            
            update_data.update({"updated_at": datetime.utcnow()})
            for field in object.__dict__:
                if field in update_data:
                    setattr(db_object, field, update_data[field])
            db.add(db_object)
            await db.commit()

        return db_object

    async def db_delete(
            self,
            db: AsyncSession,
            db_object: ModelType | None = None,
            **kwargs
    ):
        db_object = db_object or await self.get(db=db, **kwargs)
        await db.delete(db_object)
        await db.commit()

        return db_object

    async def delete(
            self,
            db: AsyncSession,
            db_object: ModelType | None = None,
            **kwargs
    ) -> ModelType | None:
        db_object = db_object or await self.get(db=db, **kwargs)
        if db_object:
            if "is_deleted" in db_object.__dict__.keys():
                object_dict = {
                    "is_deleted": True,
                    "deleted_at": datetime.utcnow()
                }
                query = update(self._model) \
                    .filter_by(**kwargs) \
                    .values(object_dict)
                
                await db.execute(query)
                await db.commit()
                await db.refresh(db_object)
            else:
                db_object = await self.db_delete(db=db, db_object=db_object, **kwargs)

        return db_object

src/app/crud/crud_base.py
igormagalhaesr/FastAPI-boilerplate
from app.crud.crud_base import CRUDBase
from app.models.post import Post
from app.schemas.post import PostCreateInternal, PostUpdate, PostUpdateInternal, PostDelete

CRUDPost = CRUDBase[Post, PostCreateInternal, PostUpdate, PostUpdateInternal, PostDelete]
crud_posts = CRUDPost(Post)

src/app/crud/crud_posts.py
igormagalhaesr/FastAPI-boilerplate
from app.crud.crud_base import CRUDBase
from app.models.user import User
from app.schemas.user import UserCreateInternal, UserUpdate, UserUpdateInternal, UserDelete

CRUDUser = CRUDBase[User, UserCreateInternal, UserUpdate, UserUpdateInternal, UserDelete]
crud_users = CRUDUser(User)

src/app/crud/crud_users.py
igormagalhaesr/FastAPI-boilerplate
from fastapi import FastAPI

from app.core.database import Base

from app.api import router
from app.core.database import async_engine as engine
from app.core.config import settings

async def create_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


def create_application() -> FastAPI:
    application = FastAPI(
        title=settings.APP_NAME,
        description=settings.APP_DESCRIPTION,
        contact=settings.CONTACT,
        license_info=settings.CONTACT
    )

    application.include_router(router)
    application.add_event_handler("startup", create_tables)

    return application


app = create_application()

src/app/main.py
igormagalhaesr/FastAPI-boilerplate
import uuid as uuid_pkg
from datetime import datetime

from sqlalchemy import String, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.core.database import Base

class Post(Base):
    __tablename__ = "post"

    id: Mapped[int] = mapped_column(
        "id", autoincrement=True, nullable=False, unique=True, primary_key=True, init=False
    )
    created_by_user_id: Mapped[int] = mapped_column(ForeignKey("user.id"))
    title: Mapped[str] = mapped_column(String(30))
    text: Mapped[str] = mapped_column(String(63206))
    uuid: Mapped[uuid_pkg.UUID] = mapped_column(
        default_factory=uuid_pkg.uuid4, primary_key=True, unique=True
    )    
    media_url: Mapped[str | None] = mapped_column(String, default=None)

    user: Mapped["User"] = relationship(back_populates="posts", lazy="selectin", init=False)

    created_at: Mapped[datetime] = mapped_column(
        DateTime, default_factory=datetime.utcnow
    )
    updated_at: Mapped[datetime | None] = mapped_column(default=None)
    deleted_at: Mapped[datetime | None] = mapped_column(default=None)
    is_deleted: Mapped[bool] = mapped_column(default=False)

src/app/models/post.py
igormagalhaesr/FastAPI-boilerplate
from typing import Optional, List
import uuid as uuid_pkg
from datetime import datetime

from sqlalchemy import String, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.core.database import Base
from app.models.post import Post

class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(
        "id", autoincrement=True, nullable=False, unique=True, primary_key=True, init=False
    )
    
    name: Mapped[str] = mapped_column(String(30))
    username: Mapped[str] = mapped_column(String(20), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String)

    profile_image_url: Mapped[str] = mapped_column(String, default="https://profileimageurl.com")
    uuid: Mapped[uuid_pkg.UUID] = mapped_column(
        default_factory=uuid_pkg.uuid4, primary_key=True, unique=True
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default_factory=datetime.utcnow
    )
    updated_at: Mapped[Optional[datetime]] = mapped_column(default=None)
    deleted_at: Mapped[Optional[datetime]] = mapped_column(default=None)
    is_deleted: Mapped[bool] = mapped_column(default=False)
    is_superuser: Mapped[bool] = mapped_column(default=False)

    posts: Mapped[List[Post]] = relationship(back_populates="user", cascade="all, delete", lazy="selectin", default_factory=list)

src/app/models/user.py
igormagalhaesr/FastAPI-boilerplate
from typing import Annotated, Optional
from datetime import datetime

from pydantic import BaseModel, Field, ConfigDict

from app.core.models import UUIDModel, TimestampModel, PersistentDeletion

class PostBase(BaseModel):
    title: Annotated[
        str,
        Field(min_length=2, max_length=30, examples=["This is my post"])
    ]
    text: Annotated[
        str, 
        Field(min_length=1, max_length=63206, examples=["This is the content of my post."])
    ]
    

class Post(TimestampModel, PostBase, UUIDModel, PersistentDeletion):
    media_url: Annotated[
        str | None, 
        Field(
            pattern=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$",
            examples=["https://www.postimageurl.com"],
            default=None
        ),
    ]
    created_by_user_id: int


class PostRead(BaseModel):
    id: int
    title: Annotated[
        str,
        Field(min_length=2, max_length=30, examples=["This is my post"])
    ]
    text: Annotated[
        str, 
        Field(min_length=1, max_length=63206, examples=["This is the content of my post."])
    ]
    media_url: Annotated[
        str | None, 
        Field(
            examples=["https://www.postimageurl.com"],
            default=None
        ),
    ]
    created_by_user_id: int
    created_at: datetime


class PostCreate(PostBase):
    model_config = ConfigDict(extra='forbid')
    
    media_url: Annotated[
        str | None, 
        Field(
            pattern=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$",
            examples=["https://www.postimageurl.com"],
            default=None
        ),
    ]


class PostCreateInternal(PostCreate):
    created_by_user_id: int


class PostUpdate(PostBase):
    model_config = ConfigDict(extra='forbid')
    
    title: Annotated[
        str | None,
        Field(
            min_length=2, 
            max_length=30, 
            examples=["This is my updated post"],
            default=None
        )
    ]
    text: Annotated[
        str | None,
        Field(
            min_length=1, 
            max_length=63206, 
            examples=["This is the updated content of my post."],
            default=None
        )
    ]
    media_url: Annotated[
        str | None,
        Field(
            pattern=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$",
            examples=["https://www.postimageurl.com"],
            default=None
        )
    ]


class PostUpdateInternal(PostUpdate):
    updated_at: datetime


class PostDelete(BaseModel):
    model_config = ConfigDict(extra='forbid')

    is_deleted: bool
    deleted_at: datetime

src/app/schemas/post.py
igormagalhaesr/FastAPI-boilerplate
from typing import Annotated, Optional
from datetime import datetime

from pydantic import BaseModel, EmailStr, Field, HttpUrl, ConfigDict

from app.core.models import UUIDModel, TimestampModel, PersistentDeletion

class UserBase(BaseModel):
    name: Annotated[
        str, 
        Field(min_length=2, max_length=30, examples=["User Userson"])
    ]
    username: Annotated[
        str, 
        Field(min_length=2, max_length=20, pattern=r"^[a-z0-9]+$", examples=["userson"])
    ]
    email: Annotated[
        EmailStr, 
        Field(examples=["user.userson@example.com"])
    ]


class User(TimestampModel, UserBase, UUIDModel, PersistentDeletion):
    profile_image_url: Annotated[
        str, 
        Field(default="https://www.profileimageurl.com")
    ]
    hashed_password: str
    is_superuser: bool = False


class UserRead(BaseModel):
    id: int
    name: Annotated[
        str, 
        Field(min_length=2, max_length=30, examples=["User Userson"])
    ]
    username: Annotated[
        str, 
        Field(min_length=2, max_length=20, pattern=r"^[a-z0-9]+$", examples=["userson"])
    ]
    profile_image_url: str


class UserCreate(UserBase):
    model_config = ConfigDict(extra='forbid')

    password: Annotated[
        str, 
        Field(pattern=r"^.{8,}|[0-9]+|[A-Z]+|[a-z]+|[^a-zA-Z0-9]+$", examples=["Str1ngst!"])
    ]


class UserCreateInternal(UserBase):
    hashed_password: str


class UserUpdate(BaseModel):
    model_config = ConfigDict(extra='forbid')

    name: Annotated[
        Optional[str], 
        Field(
            min_length=2, 
            max_length=30, 
            examples=["User Userberg"],
            default=None
        )
    ]
    username: Annotated[
        Optional[str], 
        Field(
            min_length=2, 
            max_length=20, 
            pattern=r"^[a-z0-9]+$", 
            examples=["userberg"],
            default=None
        )
    ]
    email: Annotated[
        Optional[EmailStr],
        Field(
            examples=["user.userberg@example.com"],
            default=None
        )
    ]
    profile_image_url: Annotated[
        Optional[str],
        Field(
            pattern=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$",
            examples=["https://www.profileimageurl.com"],
            default=None
        )
    ]


class UserUpdateInternal(UserUpdate):
    updated_at: datetime


class UserDelete(BaseModel):
    model_config = ConfigDict(extra='forbid')

    is_deleted: bool
    deleted_at: datetime


class UserRestoreDeleted(BaseModel):
    is_deleted: bool

src/app/schemas/user.py
igormagalhaesr/FastAPI-boilerplate
import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context

from app.core.database import Base
from app.core.config import settings
from app.models.user import *

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

config.set_main_option("sqlalchemy.url", f"postgresql+asyncpg://{settings.POSTGRES_URI}")

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """In this scenario we need to create an Engine
    and associate a connection with the context.

    """

    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""

    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()
src/migrations/env.py
igormagalhaesr/FastAPI-boilerplate
import asyncio
from sqlalchemy import select

from app.core.database import AsyncSession, local_session
from app.core.config import settings
from app.models.user import User
from app.core.security import get_password_hash

async def create_first_user(session: AsyncSession) -> None:
    name = settings.ADMIN_NAME
    email = settings.ADMIN_EMAIL
    username = settings.ADMIN_USERNAME
    hashed_password = get_password_hash(settings.ADMIN_PASSWORD)
    
    query = select(User).where(User.email == email)
    result = await session.execute(query)
    user = result.scalar_one_or_none()
    
    if user is None:
        session.add(
            User(
                name=name,
                email=email,
                username=username,
                hashed_password=hashed_password, 
                is_superuser=True
            )
        )
        
        await session.commit()

async def main():
    async with local_session() as session:
        await create_first_user(session)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

src/scripts/create_first_superuser.py
igormagalhaesr/FastAPI-boilerplate
import pytest
from fastapi.testclient import TestClient

from src.app.main import app

@pytest.fixture(scope="session")
def client():
    with TestClient(app) as _client:
        yield _client

src/tests/conftest.py
igormagalhaesr/FastAPI-boilerplate
from typing import List
import re

from fastapi.testclient import TestClient
from fastapi.routing import APIRoute

from app.main import app

def _get_token(username: str, password: str, client: TestClient):
    return client.post(
        "/api/v1/login",
        data={
            "username": username,
            "password": password
        },
        headers={"content-type": "application/x-www-form-urlencoded"}
    )

src/tests/helper.py
igormagalhaesr/FastAPI-boilerplate
from fastapi.testclient import TestClient
from decouple import config

from src.app.main import app
from tests.helper import _get_token

test_name = config("TEST_NAME")
test_username = config("TEST_USERNAME")
test_email = config("TEST_EMAIL")
test_password = config("TEST_PASSWORD")

admin_username = config("ADMIN_USERNAME")
admin_password = config("ADMIN_PASSWORD")

client = TestClient(app)

def test_post_user(client: TestClient) -> None:
    response = client.post(
        "/api/v1/user",
        json = {
            "name": test_name,
            "username": test_username,
            "email": test_email,
            "password": test_password
        }
    )
    assert response.status_code == 201

def test_get_user(client: TestClient) -> None:
    response = client.get(
        f"/api/v1/user/{test_username}"
    )
    assert response.status_code == 200

def test_get_multiple_users(client: TestClient) -> None:
    response = client.get(
        "/api/v1/users"
    )
    assert response.status_code == 200

def test_update_user(client: TestClient) -> None:
    token = _get_token(
        username=test_username, 
        password=test_password, 
        client=client
    )
    
    response = client.patch(
        f"/api/v1/user/{test_username}",
        json={
            "name": f"Updated {test_name}"
        },
        headers={"Authorization": f'Bearer {token.json()["access_token"]}'}
    )
    assert response.status_code == 200

def test_delete_user(client: TestClient) -> None:
    token = _get_token(
        username=test_username, 
        password=test_password, 
        client=client
    )

    response = client.delete(
        f"/api/v1/user/{test_username}",
        headers={"Authorization": f'Bearer {token.json()["access_token"]}'}
    )
    assert response.status_code == 200

def test_delete_db_user(client: TestClient) -> None:
    token = _get_token(
        username=admin_username, 
        password=admin_password, 
        client=client
    )

    response = client.delete(
        f"/api/v1/db_user/{test_username}",
        headers={"Authorization": f'Bearer {token.json()["access_token"]}'}
    )
    assert response.status_code == 200

src/tests/test_user.py
igormagalhaesr/FastAPI-boilerplate