Skip to content

Server Side Software Design

Jump to bottom
James Brucker edited this page Nov 22, 2025 · 4 revisions

Software Stack for Back-End

The "back-end" provides a web service, authentication, and data persistence.

I compared Javascript and Python for the back-end web service and persistence services.

  • Javascript requires many more packages and has longer learning curve
  • Javascript also seems to pose more security risks, in part due to larger number of packages required

So, I chose Python for the back-end.

I compared several Python frameworks. Top contenders were Django RestFramework (DRF), Flask, and FastAPI.

I selected FastAPI for:

  • full async support
  • high throughput and low latency
  • flexibility
  • excellent documentation. Django and Flask also have excellent documentation, but DRF documentation seems incomplete.

The drawback of FastAPI is more code for persistence (ORM), authentication, and REST route handlers. DRF provides all of these with minimal code.

Web Service URLs

The FastAPI router code (in backend/app/routers) and online documentation (http://localhost:8000/docs/) are the authoritative source for routes. As of August 2025:

Route Meaning
/api/sources Get data sources owned by current user, create a new data source
/api/sources/{id} Get, update, or delete a datasource
/api/sources/{id}/readings Routes for data source readings, including create, get, update, delete
/api/users Get users, add a new user.
/api/auth Authentication related.
/api/auth/login Authenticate using data in JSON payload.
/api/auth/password Change password. POST form-encoded data or PUT JSON format data.
/api/auth/validate Validate an access token.

URLs for Web Browsers

Route Meaning
/login GET a Login form or POST form data to login. Required by OAuth Password flow.

Project Structure for Back-end Service

Using FastAPI, SqlAlchemy for ORM, and Pydantic.

Packages have __init__.py files that are not shown here.

.env                            # Env variable definitions. Not committed to Git.
sample.env                      # Sample .env, with explanations.
backend/
├── app/
│   ├── routers/                # REST handlers for endpoints
│   │   ├── v1/                 # API versioning (not implemented yet)                 
│   │   │   ├── data_sources.py # API handlers for "data source" endpoints, e.g. an electric meter
|   │   │   ├── locations.py
│   │   │   ├── readings.py
│   │   │   ├── users.py
│   │   │   └── auth.py
│   ├── core/
│   │   ├── config.py           # App settings. Provides access to env variables and other app settings.
│   │   ├── database.py         # Database connection and session management
│   │   └── security.py         # Auth utilities
│   ├── data_access/            # Data access operations. Other possible names: persistence, storage
│   │   ├── base_dao.py         # Common code for DAO methods
│   │   ├── data_source_dao.py 
│   │   ├── ...
│   │   ├── 
│   │   └── users_dao.py
│   ├── models/                 # ORM models. Initially all models in a single file (models.py)
│   │   ├── data_source.py 
│   │   ├── ...
│   │   └── user.py
│   ├── schemas/                # Pydantic Schema for models, for validation and serialization
|   |   |                       # Like models, initially these are in a single file schemas.py
│   │   ├── data_source.py      # Too much name confusion.
│   │   └── user.py
│   ├── services/               # Business logic
│   │   ├── home_service.py
│   │   └── sensor_service.py
│   ├── utils/
│   │   ├── jwt.py              # JWT token creation, validation, and data extraction
│   │   └── validators.py       # (Optional) Custom Pydantic validators. May be in schemas
│   └── main.py                 # FastAPI app initialization
├── tests/
│   ├── conftest.py             # Session level test configuration, i.e. startup and shutdown
│   ├── fixtures.py             # Test fixtures, available in tests using dependency injection
│   └── test_*.py               # Unit tests
├── migrations/                 # Alembic migrations
├── requirements.txt            # Required Python packages for run-time
├── requirements-dev.txt        # Includes `requirements.txt` plus packages for testing and dev.
├── README.md                   # README for the backend service
└── setup.cfg                   # Old-style "conf" file for flake8 configuration & possibly more (pytest).

Key Files Explained

1. Database Layer and ORM

Using SqlAlchemy ORM for persistence operations. All models should extend a common Base class, which is defined in core/database.py since it needs the database engine to perform any schema operations.

For models I use the SqlAlchemy 2.0 syntax, including Mapped[type] and mapped_column, which differs from syntax used in most tutorials and examples.

For primary keys, I considered using UUID, especially chronologically ordered UUID (UUIDv7). But most classes would not see a real benefit and UUID requires 16 bytes compared with 4 bytes for an Integer id.

For measurements, I plan to use UUID so that measurements can be recorded off-line on Android devices, without risk of id collision.

"""User model"""
from datetime import datetime
from sqlalchemy import String, Integer, TIMESTAMP
from sqlalchemy.orm import Mapped, mapped_column, relationship
# database knows about the engine, so let it create a shared Base class
from app.core.database import Base, db
from app.config.config import MAX_EMAIL, MAX_NAME

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, nullable=False)
    email: Mapped[str] = mapped_column(String(MAX_EMAIL), unique=True, nullable=False)
    username: Mapped[str] = mapped_column(String(MAX_NAME))
    created_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False, 
                                                 server_default=func.now())

Using server-side operations like server_default=func.now() with async code throws exceptions. This is a known issue. So in actual code I replace server-side operations with client-side such as:

def utcnow() -> datetime:
    """Return the current datetime as a timezone aware value."""
    return datetime.now(timezone.utc)

class User(Base):
    ...
    created_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False, 
                                                 default=utc_now)

2. Pydantic Schemas

Pydantic schemas perform data validation and provide some serialization/deserialization to/from JSON. FastAPI makes considerable use of this in REST API endpoints, which eliminates a lot of routine serialization code. For example, a schema object can be used a parameter and FastAPI will automatically populate it from the request body, or the return type can be specified as a schema class and FastAPI will automatically construct the a JSON or XML return body using the data.

I use the Pydantic v2 syntax for model_config instead of Config inner class.

class UserCreate(BaseModel):
    """User attributes that are specified to create a new User entity."""
    email: EmailStr
    username: Optional[str] = None

class User(UserCreate):
    """The complete User schema."""
    id: int
    # In model classes, these default to current time
    created_at: datetime = datetime.now(timezone.utc)
    updated_at: datetime = datetime.now(timezone.utc)
    # model_config replaces the Config inner-class in Pydantic 2.0
    model_config = ConfigDict(from_attributes=True)

3. Persistence (CRUD) Operations

I decided to use async code for ORM persistence operations because it offers 2X - 10X higher throughput when used with FastAPI (according to claims).
Another motivating factor is deployment: the database may be on a separate (virtual) server and in any event is containerized, so network latency may offer significant delay. Async code is a good way to avoid tying up threads of execution waiting for database operations.

Initially I am not using classes for CRUD methods -- just plain functions. May change this if the class-based approach offers opportunity for code reuse.

This decision has a big effect on the code. You need specialized database drivers along with the corresponding URL, e.g. postgresql+async://hostname/db_name). Even the unit test code is affected.

In CRUD functions, the input is a schema object and any returned entities are model object(s). Deepseek recommended that return values be schema objects. I think its too much coupling (trying to guess what the caller wants to do with the object) and unnecessary overhead (converting models returned by SqlAlchemy into schema instances).

async def create_user(session: AsyncSession, user_data: schemas.UserCreate) -> models.User:
    """Add a new user to persistent storage, assigning a user.id and creation date.

    :param session: database connection "session" object
    :param user_data: schema object containing attributes for a new user entity
    :returns: a User model object populated with values of the corresponding User entity
    :raises IntegrityError: if uniqueness constraint(s) violated
    :raises ValueError: if any required values are missing or invalid
    """
    user = models.User(**user_data.model_dump())
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

4. API Endpoints (FastAPI route handlers)

The API endpoints try to follow REST design guidelines, and implemented using FastAPI.

Depends performs dependency injection. Most of the example code on the Internet uses misleading names for the database session. There are mountains of code like this:

   db: AsyncSession = Depends(get_db)

The type hint (AsyncSession) is a clue that this is bogus.

from fastapi import APIRouter, Depends, HTTPException
from app import schemas
from app.core.database import get_session
from app.data_access import user_dao

router = APIRouter(prefix="/api/v1/measurements", tags=["measurements"])

@router.post("/users", response_model=schemas.User)
async def create_user(user: schemas.UserCreate, session: AsyscSession = Depends(get_session)):
    user = await user_dao.create_user(session, obj_in=measurement)
    return user

5. Service Layer (app/services/sensor_service.py)

This code provided by a coding Chatbot. I haven't tried it yet.

from app.db import crud, schemas
from app.db.session import get_db

class SensorService:
    def __init__(self):
        self.crud = crud.CRUDMeasurement()

    async def add_measurement(
        self, measurement: schemas.MeasurementCreate, db: AsyncSession
        ) -> schemas.MeasurementResponse:
        # Add business logic (e.g., validate sensor exists)
        return await self.crud.create(db, obj_in=measurement)

Design Principles

According to the code generating Chat-bots, this code shows:

  1. Separation of Concerns:

    • models.py: Pure database schema (SQLAlchemy)
    • schemas.py: API data contracts (Pydantic)
    • user_dao.py: Database operations
    • services/: Business logic
    • routers/: Route handling

  2. Async First:

    • Use asyncpg + SQLAlchemy 2.0 async sessions
    • All endpoints are async def

  3. Validation Layers:

    • Pydantic: Validates API input/output
    • SQLAlchemy: Enforces database constraints

  4. Testability:

    • Isolated service layer for easy mocking
    • Example test: 
      async def test_create_measurement(async_client):
    response = await async_client.post(
        "/homelog/v1/measurements/",
        json={"sensor_id": 1, "value": 42.5}
    )
    assert response.status_code == 200
    assert response.json()["value"] == 42.5

This structure promotes:

  • Maintainability: Clear boundaries between components
  • Scalability: Ready for distributed systems
  • Consistency: Matches your exact database schema
  • Documentation: Auto-generated OpenAPI specs (/docs endpoint)

Performance Comparison of DjangoRestFramework and FastAPI

Using AWS and Supabase with this benchmark setup:

  • Python 3.10
  • Machine: AWS t3.medium (2 vCPU, 4GB RAM)
  • Database: PostgreSQL (Supabase free tier)
  • Test Tool: locust (simulates 10–100 concurrent users)
  • Endpoint: Fetch residential data for a home (/usage/{home_id})

1. Performance Results

Metric FastAPI (Async) Django REST Framework (Sync)
Avg. Latency (10 users) 12 ms 28 ms
Avg. Latency (100 users) 45 ms 210 ms
Max Throughput (req/sec) 1,200 350
Cold Start Time ~300ms (with ASGI) ~600ms (WSGI)
Memory Usage 55 MB (lightweight) 110 MB (ORM overhead)
Concurrency Support Async (ASGI) Sync (WSGI)

Concurrent Users Test (100 Concurrent Users)

  • FastAPI: 1,100 requests/sec with 95% latency < 100ms.
  • DRF: Peak 320 requests/sec with 95% latency < 300ms.

2. Latency & Throughput

  • FastAPI is 2–5x faster under load due to:
    • Async I/O (non-blocking database queries).
    • No ORM overhead (uses asyncpg or SQLAlchemy Core).
  • Django REST Framework slows down at scale due to:
    • Synchronous ORM (blocking DB calls).
    • Heavy middleware stack.

3. Cold Starts (Serverless)

  • FastAPI wins in Vercel/Railway:
  • Smaller Docker images (~150MB vs Django’s ~400MB).
  • ASGI servers (Uvicorn) start faster than WSGI (Gunicorn).

4. Development Speed

  • DRF wins for rapid prototyping:
  • DRF has built-in admin panel, auth, and serializers.
  • Less boilerplate for CRUD.
  • FastAPI requires more manual setup but offers flexibility.

5. Database Queries

  • FastAPI + SQLAlchemy Core: ~8 ms/query (async).
  • DRF + Django ORM: ~22 ms/query (sync).

6. Optimizing Each Framework

Suggestions from the AI chatbots, but I may not implement them.

FastAPI

  1. Use asyncpg instead of SQLAlchemy ORM.
  2. Enable GZip compression: 
    from fastapi.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware)
  3. Cache responses with @lru_cache for static data.

Django REST Framework

  1. Use django-cacheops to cache ORM queries: 
    @cacheops.cached(timeout=60)
def get_usage(home_id):
    return PowerUsage.objects.get(home_id=home_id)
  2. Use uvicorn (ASGI) for partial async support.

Clone this wiki locally