migrated to sqlmodel

Author: igorbenav

README.md

@@ -42,6 +42,9 @@
 - [`Docker Compose`](https://docs.docker.com/compose/) With a single command, create and start all the services from your configuration.
 - [`NGINX`](https://nginx.org/en/) High-performance low resource consumption web server used for Reverse Proxy and Load Balancing.
 
+> \[!TIP\] 
+> If you want the `SQLAlchemy + Pydantic` version instead, head to [Fastapi-boilerplate](https://github.com/igorbenav/FastAPI-boilerplate).
+
 ## 1. Features
 
 - ⚡️ Fully async

pyproject.toml

@@ -31,6 +31,7 @@ arq = "^0.25.0"
 gunicorn = "^22.0.0"
 bcrypt = "^4.1.1"
 fastcrud = "^0.12.0"
+sqlmodel = "^0.0.18"
 
 
 [build-system]

src/app/api/dependencies.py

@@ -13,7 +13,7 @@
 from ..crud.crud_tier import crud_tiers
 from ..crud.crud_users import crud_users
 from ..models.user import User
-from ..schemas.rate_limit import sanitize_path
+from ..models.rate_limit import sanitize_path
 
 logger = logging.getLogger(__name__)
 

src/app/api/v1/posts.py

@@ -10,8 +10,8 @@
 from ...core.utils.cache import cache
 from ...crud.crud_posts import crud_posts
 from ...crud.crud_users import crud_users
-from ...schemas.post import PostCreate, PostCreateInternal, PostRead, PostUpdate
-from ...schemas.user import UserRead
+from ...models.post import PostCreate, PostCreateInternal, PostRead, PostUpdate
+from ...models.user import UserRead
 
 router = APIRouter(tags=["posts"])
 

src/app/api/v1/rate_limits.py

@@ -9,7 +9,7 @@
 from ...core.exceptions.http_exceptions import DuplicateValueException, NotFoundException, RateLimitException
 from ...crud.crud_rate_limit import crud_rate_limits
 from ...crud.crud_tier import crud_tiers
-from ...schemas.rate_limit import RateLimitCreate, RateLimitCreateInternal, RateLimitRead, RateLimitUpdate
+from ...models.rate_limit import RateLimitCreate, RateLimitCreateInternal, RateLimitRead, RateLimitUpdate
 
 router = APIRouter(tags=["rate_limits"])
 

src/app/api/v1/tasks.py

@@ -5,7 +5,7 @@
 
 from ...api.dependencies import rate_limiter
 from ...core.utils import queue
-from ...schemas.job import Job
+from ...models.job import Job
 
 router = APIRouter(prefix="/tasks", tags=["tasks"])
 

src/app/api/v1/tiers.py

@@ -8,7 +8,7 @@
 from ...core.db.database import async_get_db
 from ...core.exceptions.http_exceptions import DuplicateValueException, NotFoundException
 from ...crud.crud_tier import crud_tiers
-from ...schemas.tier import TierCreate, TierCreateInternal, TierRead, TierUpdate
+from ...models.tier import TierCreate, TierCreateInternal, TierRead, TierUpdate
 
 router = APIRouter(tags=["tiers"])
 

src/app/api/v1/users.py

@@ -11,9 +11,8 @@
 from ...crud.crud_rate_limit import crud_rate_limits
 from ...crud.crud_tier import crud_tiers
 from ...crud.crud_users import crud_users
-from ...models.tier import Tier
-from ...schemas.tier import TierRead
-from ...schemas.user import UserCreate, UserCreateInternal, UserRead, UserTierUpdate, UserUpdate
+from ...models.tier import Tier, TierRead
+from ...models.user import UserCreate, UserCreateInternal, UserRead, UserTierUpdate, UserUpdate
 
 router = APIRouter(tags=["users"])
 

src/app/core/db/database.py

@@ -1,14 +1,10 @@
 from sqlalchemy.ext.asyncio import create_async_engine
 from sqlalchemy.ext.asyncio.session import AsyncSession
-from sqlalchemy.orm import DeclarativeBase, MappedAsDataclass, sessionmaker
+from sqlalchemy.orm import sessionmaker
 
 from ..config import settings
 
 
-class Base(DeclarativeBase, MappedAsDataclass):
-    pass
-
-
 DATABASE_URI = settings.POSTGRES_URI
 DATABASE_PREFIX = settings.POSTGRES_ASYNC_PREFIX
 DATABASE_URL = f"{DATABASE_PREFIX}{DATABASE_URI}"

src/app/core/db/token_blacklist.py

@@ -1,14 +1,7 @@
 from datetime import datetime
+from sqlmodel import SQLModel, Field
 
-from sqlalchemy import DateTime, String
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .database import Base
-
-
-class TokenBlacklist(Base):
-    __tablename__ = "token_blacklist"
-
-    id: Mapped[int] = mapped_column("id", autoincrement=True, nullable=False, unique=True, primary_key=True, init=False)
-    token: Mapped[str] = mapped_column(String, unique=True, index=True)
-    expires_at: Mapped[datetime] = mapped_column(DateTime)
+class TokenBlacklist(SQLModel, table=True):
+    id: int = Field(default=None, primary_key=True, nullable=False)
+    token: str = Field(index=True, nullable=False, unique=True)
+    expires_at: datetime = Field(nullable=False)

src/app/core/setup.py

@@ -10,6 +10,7 @@
 from fastapi import APIRouter, Depends, FastAPI
 from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
 from fastapi.openapi.utils import get_openapi
+from sqlmodel import SQLModel
 
 from ..api.dependencies import get_current_superuser
 from ..middleware.client_cache_middleware import ClientCacheMiddleware
@@ -24,15 +25,14 @@
     RedisRateLimiterSettings,
     settings,
 )
-from .db.database import Base
 from .db.database import async_engine as engine
 from .utils import cache, queue, rate_limit
 
 
 # -------------- database --------------
 async def create_tables() -> None:
     async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
+        await conn.run_sync(SQLModel.metadata.create_all)
 
 
 # -------------- cache --------------

src/app/core/utils/rate_limit.py

@@ -4,7 +4,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ...core.logger import logging
-from ...schemas.rate_limit import sanitize_path
+from ...models.rate_limit import sanitize_path
 
 logger = logging.getLogger(__name__)
 

src/app/crud/crud_posts.py

@@ -1,7 +1,6 @@
 from fastcrud import FastCRUD
 
-from ..models.post import Post
-from ..schemas.post import PostCreateInternal, PostDelete, PostUpdate, PostUpdateInternal
+from ..models.post import Post, PostCreateInternal, PostDelete, PostUpdate, PostUpdateInternal
 
 CRUDPost = FastCRUD[Post, PostCreateInternal, PostUpdate, PostUpdateInternal, PostDelete]
 crud_posts = CRUDPost(Post)

src/app/crud/crud_rate_limit.py

@@ -1,7 +1,6 @@
 from fastcrud import FastCRUD
 
-from ..models.rate_limit import RateLimit
-from ..schemas.rate_limit import RateLimitCreateInternal, RateLimitDelete, RateLimitUpdate, RateLimitUpdateInternal
+from ..models.rate_limit import RateLimit, RateLimitCreateInternal, RateLimitDelete, RateLimitUpdate, RateLimitUpdateInternal
 
 CRUDRateLimit = FastCRUD[RateLimit, RateLimitCreateInternal, RateLimitUpdate, RateLimitUpdateInternal, RateLimitDelete]
 crud_rate_limits = CRUDRateLimit(RateLimit)

src/app/crud/crud_tier.py

@@ -1,7 +1,6 @@
 from fastcrud import FastCRUD
 
-from ..models.tier import Tier
-from ..schemas.tier import TierCreateInternal, TierDelete, TierUpdate, TierUpdateInternal
+from ..models.tier import Tier, TierCreateInternal, TierDelete, TierUpdate, TierUpdateInternal
 
 CRUDTier = FastCRUD[Tier, TierCreateInternal, TierUpdate, TierUpdateInternal, TierDelete]
 crud_tiers = CRUDTier(Tier)

src/app/crud/crud_users.py

@@ -1,7 +1,6 @@
 from fastcrud import FastCRUD
 
-from ..models.user import User
-from ..schemas.user import UserCreateInternal, UserDelete, UserUpdate, UserUpdateInternal
+from ..models.user import User, UserCreateInternal, UserDelete, UserUpdate, UserUpdateInternal
 
 CRUDUser = FastCRUD[User, UserCreateInternal, UserUpdate, UserUpdateInternal, UserDelete]
 crud_users = CRUDUser(User)

src/app/models/job.py

@@ -0,0 +1,5 @@
+from sqlmodel import SQLModel
+
+
+class Job(SQLModel):
+    id: str

src/app/models/post.py

@@ -1,23 +1,49 @@
-import uuid as uuid_pkg
-from datetime import UTC, datetime
+from datetime import datetime
+from typing import Optional
+from sqlmodel import SQLModel, Field, Relationship
+from uuid import uuid4
 
-from sqlalchemy import DateTime, ForeignKey, String
-from sqlalchemy.orm import Mapped, mapped_column
 
-from ..core.db.database import Base
+class PostBase(SQLModel):
+    title: str = Field(..., min_length=2, max_length=30, schema_extra={"example": "This is my post"})
+    text: str = Field(..., min_length=1, max_length=63206, schema_extra={"example": "This is the content of my post."})
 
 
-class Post(Base):
-    __tablename__ = "post"
+class Post(PostBase, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    created_by_user_id: int = Field(foreign_key="user.id")
+    media_url: Optional[str] = Field(default=None, regex=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$", schema_extra={"example": "https://www.postimageurl.com"})
+    created_at: datetime = Field(default_factory=lambda: datetime.now(datetime.timezone.utc))
+    updated_at: Optional[datetime] = None
+    deleted_at: Optional[datetime] = None
+    is_deleted: bool = Field(default=False)
 
-    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"), index=True)
-    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)
 
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default_factory=lambda: datetime.now(UTC))
-    updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
-    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
-    is_deleted: Mapped[bool] = mapped_column(default=False, index=True)
+class PostRead(PostBase):
+    id: int
+    created_by_user_id: int
+    media_url: Optional[str]
+    created_at: datetime
+
+
+class PostCreate(PostBase):
+    media_url: Optional[str] = Field(default=None, regex=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$", schema_extra={"example": "https://www.postimageurl.com"})
+
+
+class PostCreateInternal(PostCreate):
+    created_by_user_id: int
+
+
+class PostUpdate(SQLModel):
+    title: Optional[str] = Field(default=None, min_length=2, max_length=30)
+    text: Optional[str] = Field(default=None, min_length=1, max_length=63206)
+    media_url: Optional[str] = Field(default=None, regex=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$")
+
+
+class PostUpdateInternal(PostUpdate):
+    updated_at: Optional[datetime] = Field(default_factory=datetime.utcnow)
+
+
+class PostDelete(SQLModel):
+    is_deleted: bool
+    deleted_at: datetime

src/app/models/rate_limit.py

@@ -1,20 +1,58 @@
-from datetime import UTC, datetime
+from datetime import datetime
+from typing import Optional
+from sqlmodel import SQLModel, Field
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String
-from sqlalchemy.orm import Mapped, mapped_column
 
-from ..core.db.database import Base
+def sanitize_path(path: str) -> str:
+    return path.strip("/").replace("/", "_")
 
 
-class RateLimit(Base):
-    __tablename__ = "rate_limit"
+class RateLimitBase(SQLModel):
+    path: str = Field(..., schema_extra={"example": "users"})
+    limit: int = Field(..., schema_extra={"example": 5})
+    period: int = Field(..., schema_extra={"example": 60})
 
-    id: Mapped[int] = mapped_column("id", autoincrement=True, nullable=False, unique=True, primary_key=True, init=False)
-    tier_id: Mapped[int] = mapped_column(ForeignKey("tier.id"), index=True)
-    name: Mapped[str] = mapped_column(String, nullable=False, unique=True)
-    path: Mapped[str] = mapped_column(String, nullable=False)
-    limit: Mapped[int] = mapped_column(Integer, nullable=False)
-    period: Mapped[int] = mapped_column(Integer, nullable=False)
+    @classmethod
+    def validate_path(cls, v: str) -> str:
+        return sanitize_path(v)
 
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default_factory=lambda: datetime.now(UTC))
-    updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
+
+class RateLimit(RateLimitBase, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    tier_id: int = Field(foreign_key="tier.id")
+    name: Optional[str] = Field(default=None, schema_extra={"example": "users:5:60"})
+    created_at: datetime = Field(default_factory=lambda: datetime.now(datetime.timezone.utc))
+    updated_at: Optional[datetime] = None
+
+
+class RateLimitRead(RateLimitBase):
+    id: int
+    tier_id: int
+    name: str
+
+
+class RateLimitCreate(RateLimitBase):
+    name: Optional[str] = Field(default=None, schema_extra={"example": "api_v1_users:5:60"})
+
+
+class RateLimitCreateInternal(RateLimitCreate):
+    tier_id: int
+
+
+class RateLimitUpdate(SQLModel):
+    path: Optional[str] = Field(default=None)
+    limit: Optional[int] = None
+    period: Optional[int] = None
+    name: Optional[str] = None
+
+    @classmethod
+    def validate_path(cls, v: Optional[str]) -> Optional[str]:
+        return sanitize_path(v) if v is not None else None
+
+
+class RateLimitUpdateInternal(RateLimitUpdate):
+    updated_at: Optional[datetime] = Field(default_factory=datetime.utcnow)
+
+
+class RateLimitDelete(SQLModel):
+    pass

src/app/models/tier.py

@@ -1,16 +1,38 @@
-from datetime import UTC, datetime
+from datetime import datetime
+from typing import Optional
+from sqlmodel import SQLModel, Field
 
-from sqlalchemy import DateTime, String
-from sqlalchemy.orm import Mapped, mapped_column
 
-from ..core.db.database import Base
+class TierBase(SQLModel):
+    name: str = Field(..., schema_extra={"example": "free"})
 
 
-class Tier(Base):
-    __tablename__ = "tier"
+class Tier(TierBase, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    created_at: datetime = Field(default_factory=lambda: datetime.now(datetime.timezone.utc))
+    updated_at: Optional[datetime] = None
 
-    id: Mapped[int] = mapped_column("id", autoincrement=True, nullable=False, unique=True, primary_key=True, init=False)
-    name: Mapped[str] = mapped_column(String, nullable=False, unique=True)
 
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default_factory=lambda: datetime.now(UTC))
-    updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
+class TierRead(TierBase):
+    id: int
+    created_at: datetime
+
+
+class TierCreate(TierBase):
+    pass
+
+
+class TierCreateInternal(TierCreate):
+    pass
+
+
+class TierUpdate(SQLModel):
+    name: Optional[str] = None
+
+
+class TierUpdateInternal(TierUpdate):
+    updated_at: Optional[datetime] = Field(default_factory=datetime.utcnow)
+
+
+class TierDelete(SQLModel):
+    pass