⬆️ Python 3.11.7 + Pydantic V2

Author: anthonycepeda

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.10
+FROM python:3.11.7
 
 
 

Pipfile

@@ -7,13 +7,14 @@ name = "pypi"
 fastapi = "*"
 sqlmodel = "*"
 uvicorn = "*"
+pydantic-settings = "*"
 
 [dev-packages]
 black = "*"
 pylint = "*"
 
 [requires]
-python_version = "3.10"
+python_version = "3.11.7"
 
 [pipenv]
 allow_prereleases = true

Pipfile.lock

@@ -1,24 +1,30 @@
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 
-from pydantic import BaseSettings
+from api.config import Settings
 from api.database import create_db_and_tables
 from api.public import api as public_api
 from api.utils.mock_data_generator import create_heroes_and_teams
 
 
-def create_app(settings: BaseSettings):
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    create_db_and_tables()
+    create_heroes_and_teams()
+
+    yield
+
+
+def create_app(settings: Settings):
     app = FastAPI(
         title=settings.PROJECT_NAME,
         version=settings.VERSION,
         docs_url="/",
         description=settings.DESCRIPTION,
+        lifespan=lifespan,
     )
 
-    @app.on_event("startup")
-    def on_starup():
-        create_db_and_tables()
-        create_heroes_and_teams()
-
     app.include_router(public_api)
 
     return app

api/app.py

@@ -1,17 +1,16 @@
 import os
 import secrets
 
-from pydantic import BaseSettings
-from pydantic.types import Optional
+from pydantic_settings import BaseSettings
 
 
 class Settings(BaseSettings):
     PROJECT_NAME: str = f"SQLModel API - {os.getenv('ENV', 'development').capitalize()}"
     DESCRIPTION: str = "A FastAPI + SQLModel production-ready API"
-    ENV: str
+    ENV: str = "development"
     VERSION: str = "0.1"
     SECRET_KEY: str = secrets.token_urlsafe(32)
-    DATABASE_URI: Optional[str]  # pydantic.PostgresDsn can be used for PostgresDB
+    DATABASE_URI: str = "sqlite:////Users/anth/dev/fastapi-sqlmodel/database.db"
 
     class Config:
         case_sensitive = True

api/config.py

@@ -1,10 +1,12 @@
 from fastapi import APIRouter
 
+from api.public.health import views as health
 from api.public.hero import views as heroes
 from api.public.team import views as teams
 
 api = APIRouter()
 
 
+api.include_router(health.router, prefix="/health", tags=["Health"])
 api.include_router(heroes.router, prefix="/heroes", tags=["Heroes"])
 api.include_router(teams.router, prefix="/teams", tags=["Teams"])

api/public/__init__.py

@@ -0,0 +1,13 @@
+from fastapi import Depends
+from sqlmodel import Session
+
+from api.config import settings
+from api.database import get_session
+from api.public.health.models import Health, Stats
+from api.utils.logger import logger_config
+
+logger = logger_config(__name__)
+
+
+def get_health() -> Health:
+    return Health(status="OK", environment=settings.ENV)

api/public/health/crud.py

@@ -0,0 +1,24 @@
+from enum import Enum
+
+from pydantic import BaseModel
+
+
+class Status(str, Enum):
+    OK = "OK"
+    KO = "KO"
+
+
+class Environment(str, Enum):
+    dev = "development"
+    stg = "staging"
+    prd = "production"
+
+
+class Health(BaseModel):
+    status: Status | None
+    environment: Environment | None
+
+
+class Stats(BaseModel):
+    heroes: int | None
+    teams: int | None

api/public/health/models.py

@@ -0,0 +1,20 @@
+from fastapi import APIRouter, Depends, status
+from sqlmodel import Session
+
+from api.database import Session, get_session
+from api.public.health.crud import get_health
+from api.public.health.models import Health, Stats
+from api.utils.logger import logger_config
+
+router = APIRouter()
+logger = logger_config(__name__)
+
+
+@router.get(
+    "",
+    response_model=Health,
+    status_code=status.HTTP_200_OK,
+    responses={200: {"model": Health}},
+)
+def health(db: Session = Depends(get_session)):
+    return get_health()

api/public/health/views.py

@@ -1,4 +1,3 @@
-from pydantic.types import Optional, List
 from sqlmodel import Field, Relationship, SQLModel
 
 from api.public.team.models import Team
@@ -8,10 +7,10 @@
 class HeroBase(SQLModel):
     name: str
     secret_name: str
-    age: Optional[int] = None
+    age: int | None = None
 
     class Config:
-        schema_extra = {
+        json_schema_extra = {
             "example": {
                 "id": 1,
                 "name": "Super Man",
@@ -23,8 +22,8 @@ class Config:
 
 
 class Hero(HeroBase, table=True):
-    id: Optional[int] = Field(default=None, primary_key=True)
-    teams: List[Team] = Relationship(back_populates="heroes", link_model=HeroTeamLink)
+    id: int | None = Field(default=None, primary_key=True)
+    teams: list[Team] = Relationship(back_populates="heroes", link_model=HeroTeamLink)
 
 
 class HeroCreate(HeroBase):
@@ -33,20 +32,20 @@ class HeroCreate(HeroBase):
 
 class HeroRead(HeroBase):
     id: int
-    name: Optional[str] = None
-    secret_name: Optional[str] = None
-    age: Optional[int] = None
-    teams: List[Team] = None
+    name: str | None = None
+    secret_name: str | None = None
+    age: int | None = None
+    teams: list[Team] = None
 
 
 class HeroUpdate(HeroBase):
-    name: Optional[str] = None
-    secret_name: Optional[str] = None
-    age: Optional[int] = None
-    teams: List[Team] = None
+    name: str | None = None
+    secret_name: str | None = None
+    age: int | None = None
+    teams: list[Team] = None
 
     class Config:
-        schema_extra = {
+        json_schema_extra = {
             "example": {
                 "name": "Super Man",
                 "secret_name": "Clark Kent",

api/public/hero/models.py

@@ -1,5 +1,4 @@
 from fastapi import APIRouter, Depends, Query
-from pydantic.types import List
 from sqlmodel import Session
 
 from api.database import get_session
@@ -20,7 +19,7 @@ def create_a_hero(hero: HeroCreate, db: Session = Depends(get_session)):
     return create_hero(hero=hero, db=db)
 
 
-@router.get("", response_model=List[HeroRead])
+@router.get("", response_model=list[HeroRead])
 def get_heroes(
     offset: int = 0,
     limit: int = Query(default=100, lte=100),

api/public/hero/views.py

@@ -1,4 +1,3 @@
-from pydantic.types import List, Optional
 from sqlmodel import Field, Relationship, SQLModel
 
 from api.utils.generic_models import HeroTeamLink
@@ -9,7 +8,7 @@ class TeamBase(SQLModel):
     headquarters: str
 
     class Config:
-        schema_extra = {
+        json_schema_extra = {
             "example": {
                 "name": "wonderful league",
                 "headquarters": "Fortress of Solitude",
@@ -18,9 +17,9 @@ class Config:
 
 
 class Team(TeamBase, table=True):
-    id: Optional[int] = Field(default=None, primary_key=True)
+    id: int | None = Field(default=None, primary_key=True)
 
-    heroes: List["Hero"] = Relationship(back_populates="teams", link_model=HeroTeamLink)
+    heroes: list["Hero"] = Relationship(back_populates="teams", link_model=HeroTeamLink)
 
 
 class TeamCreate(TeamBase):
@@ -29,11 +28,11 @@ class TeamCreate(TeamBase):
 
 class TeamRead(TeamBase):
     id: int
-    name: Optional[str] = None
-    headquarters: Optional[str] = None
-    heroes: List = None
+    name: str | None = None
+    headquarters: str | None = None
+    heroes: list = None
 
 
 class TeamUpdate(TeamBase):
-    name: Optional[str] = None
-    headquarters: Optional[str] = None
+    name: str | None = None
+    headquarters: str | None = None

api/public/team/models.py

@@ -1,5 +1,4 @@
 from fastapi import APIRouter, Depends, Query
-from pydantic.types import List
 from sqlmodel import Session
 
 from api.database import get_session
@@ -20,7 +19,7 @@ def create_a_team(team: TeamCreate, db: Session = Depends(get_session)):
     return create_team(team=team, db=db)
 
 
-@router.get("", response_model=List[TeamRead])
+@router.get("", response_model=list[TeamRead])
 def get_teams(
     offset: int = 0,
     limit: int = Query(default=100, lte=100),

api/public/team/views.py

@@ -1,11 +1,6 @@
-from pydantic.types import Optional
 from sqlmodel import Field, SQLModel
 
 
 class HeroTeamLink(SQLModel, table=True):
-    team_id: Optional[int] = Field(
-        default=None, foreign_key="team.id", primary_key=True
-    )
-    hero_id: Optional[int] = Field(
-        default=None, foreign_key="hero.id", primary_key=True
-    )
+    team_id: int | None = Field(default=None, foreign_key="team.id", primary_key=True)
+    hero_id: int | None = Field(default=None, foreign_key="hero.id", primary_key=True)

api/utils/generic_models.py

@@ -0,0 +1,19 @@
+import logging
+
+
+def logger_config(module):
+    """
+    Logger function. Extends Python loggin module and set a custom config.
+    params: Module Name. e.i: logger_config(__name__).
+    return: Custom logger_config Object.
+    """
+    formatter = logging.Formatter("%(asctime) %(levelname) %(message)")
+    handler = logging.StreamHandler()
+    handler.setFormatter(formatter)
+
+    custom_logger = logging.getLogger(module)
+    custom_logger.setLevel(logging.DEBUG)
+
+    custom_logger.addHandler(handler)
+
+    return custom_logger