Note: The important chunks of code are pasted in the question but the larger representation of the app is available on GitHub.
Some time ago I was tasked to create a simple FastAPI app. I did manage to pull it off and I already have it running with no errors, But I want to improve it. This project has a lot in common with a simple blog app, where an author (an operator) can write posts (alerts).
In this project I use FastAPI, SQLAlchemy, pydantic, alembic, uvicorn, pytest, and passlib
My main questions are:
- Is this app suitable for the production phase?
- Am I using industry-standard routines and functions?
- Is the code pythonic?
- Are schemas/models correct? Are their responsibilities distributed well?
- Does the app follow SOLID principles? Does it even need to (considering its small scale)?
My goals for this project are:
- be scalable with a predictable structure and readable clean code (my main goal),
- have database models/schemas that make sense, and are maintainable for possible future modifications,
- dockerize,
- connect to a React front-end and a PostgreSQL database, and
- run automated tests.
(NOTE: the actual app - fully or partially - already includes the last three items, but I welcome any suggestions.)
Future plans:
- have JWT authorization
- have roles for
operators with different permissions - run on a server
I appreciate any comments (even opinionated ones on the design of this app).
Here is my code:
main.py
from fastapi import FastAPI
from app.apis.base import api_router
from app.db.base import Base
from app.db.session import engine
app = FastAPI()
# Create all database tables
Base.metadata.create_all(bind=engine)
# Include API router
app.include_router(api_router)
@app.get("/")
def health_check():
return {"message": "Health: Check🚀"}
schemas.py
from typing import Optional
from sqlalchemy import JSON
from .operator import ShowOperator
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
class CreateAlert(BaseModel):
showAlert: bool | None = False
alertContent: str
androidVersion: str | None = ""
launcherVersion: str | None = ""
releaseDate: datetime | None = datetime.utcnow()
class ShowAlert(BaseModel):
id: int
showAlert: bool
alertContent: str
androidVersion: str
launcherVersion: str
releaseDate: datetime
lastUpdate: datetime
author: ShowOperator
author_id: int
class Config:
orm_mode = True
class UpdateAlert(CreateAlert):
lastUpdate: datetime | None = datetime.utcnow()
class CreateOperator(BaseModel):
email: EmailStr
password: str = Field(..., min_length=4)
fullName: str
phoneNumber: str
accessLevel: str | None = "audit"
createdDate: str | None = datetime.utcnow()
class ShowOperator(BaseModel):
id: int
email: EmailStr
accessLevel: str
fullName: str
phoneNumber: str
class Config:
orm_mode = True
class UpdateOperator(CreateOperator):
pass
models.py
from sqlalchemy import DateTime, Boolean, Column, ForeignKey, Integer, String, Integer
from sqlalchemy.orm import relationship
from ...db.base_class import Base
from datetime import datetime
from enum import Enum
class Alert(Base):
id = Column(Integer, primary_key=True, index=True)
showAlert = Column(Boolean, default=False)
alertContent = Column(String, nullable=False)
androidVersion = Column(String, nullable=False)
launcherVersion = Column(String, nullable=False)
releaseDate = Column(DateTime, nullable=False,)
lastUpdate = Column(DateTime)
author_id = Column(Integer, ForeignKey("operator.id"))
author = relationship("Operator", back_populates="alerts")
class AccessLevel(str, Enum):
audit = "audit"
content = "content"
manager = "manager"
class Operator(Base):
id = Column(Integer, primary_key=True, index=True)
email = Column(String, nullable=False, unique=True, index=True)
password = Column(String, nullable=False)
fullName = Column(String)
isLoggedIn = Column(Boolean, nullable=False, default=False)
phoneNumber = Column(String)
createdDate = Column(DateTime)
accessLevel = Column(String) # enum!
alerts = relationship(
"Alert", back_populates="author", cascade="all, delete")
API v1 route, alert repo functions
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from ...db.repo.alert import (
create_new_alert,
delete_alert,
retreive_alert,
retreive_all_alerts,
update_alert,
)
from ...db.session import get_db
from ...schemas.alert import CreateAlert, ShowAlert, UpdateAlert
router = APIRouter()
@router.post("/alerts", response_model=ShowAlert, status_code=status.HTTP_201_CREATED)
async def create_one_alert(alert: CreateAlert, db: Session = Depends(get_db)):
alert = create_new_alert(alert=alert, db=db, author_id=1)
return alert
@router.get("/alert/{id}", response_model=ShowAlert)
def get_one_alert(id: int, db: Session = Depends(get_db)):
alert = retreive_alert(id=id, db=db)
if not alert:
raise HTTPException(
detail=f"Alert with ID {id} does not exist.",
status_code=status.HTTP_404_NOT_FOUND,
)
return alert
@router.get("/alerts/all", response_model=List[ShowAlert])
def get_all_alerts(db: Session = Depends(get_db)):
alerts = retreive_all_alerts(db=db)
return alerts
@router.put("/alert/{id}", response_model=ShowAlert)
def update_one_alert(id: int, alert: UpdateAlert, db: Session = Depends(get_db)):
alert = update_alert(id=id, alert=alert, author_id=1, db=db)
if not alert:
raise HTTPException(
detail=f"Alert with id {id} does not exist",
tatus_code=status.HTTP_404_NOT_FOUND,
)
return alert
@router.delete("/alert/{id}")
def delete_one_alert(id: int, db: Session = Depends(get_db)):
message = delete_alert(id=id, author_id=1, db=db)
if message.get("error"):
raise HTTPException(
detail=message.get("error"), status_code=status.HTTP_400_BAD_REQUEST
)
return {"msg": f"Successfully deleted alert with id {id}"}
API v1 route, operator repo functions
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from ...db.repo.operator import (
create_new_operator,
delete_operator,
retreive_all_operators,
retreive_operator,
update_operator,
)
from ...db.session import get_db
from ...schemas.operator import CreateOperator, ShowOperator, UpdateOperator
router = APIRouter()
@router.post("/operators", response_model=ShowOperator, status_code=status.HTTP_201_CREATED)
def create_one_operator(operator: CreateOperator, db: Session = Depends(get_db)):
operator = create_new_operator(operator=operator, db=db)
return operator
@router.get("/operator/{id}", response_model=ShowOperator)
def get_one_operator(id: int, db: Session = Depends(get_db)):
operator = retreive_operator(id=id, db=db)
if not operator:
raise HTTPException(
detail=f"Operator with ID {id} does not exist.", status_code=status.HTTP_404_NOT_FOUND)
return operator
@router.get("/operators/all", response_model=List[ShowOperator])
def get_all_operators(db: Session = Depends(get_db)):
operators = retreive_all_operators(db=db)
return operators
@router.put("/operator/{id}", response_model=ShowOperator)
def update_one_operator(id: int, operator: UpdateOperator, db: Session = Depends(get_db)):
operator = update_operator(id=id, operator=operator, db=db)
if not operator:
raise HTTPException(detail=f"Operator with id {id} does not exist")
return operator
@router.delete("/operator/{id}")
def delete_one_operator(id: int, db: Session = Depends(get_db)):
message = delete_operator(id=id, db=db)
if message.get("error"):
raise HTTPException(detail=message.get("error"),
status_code=status.HTTP_400_BAD_REQUEST)
return {"msg": f"Successfully deleted operator with id {id}"}
