CRUD and Dependency Injection

This document covers Dependency Injection (DI) with Depends(), building CRUD functions using the Repository pattern, and understanding ASGI applications and selectors in FastAPI.

---

📚 Table of Contents

1. Part 1: Application Overview
2. Part 2: AsyncIO vs Synchronous Sessions
3. Part 3: Dependency Injection
4. Part 4: CRUD Operations (Repository Pattern)
5. Part 5: ASGI Applications
6. Part 6: Routes and Dependant
7. Appendix: Selectors and Event-Driven I/O

---

Part 1: Application Overview

The big picture of a FastAPI application with database integration:

Key Topics

• Dependency Injection (DI): Using Depends() to inject reusable logic
• Database Session: Creating a get_db_session dependency
• CRUD Functions: Building with Repository pattern

---

Part 2: AsyncIO vs Synchronous Sessions

Synchronous SQLAlchemy (Classic)

Operations block the Python thread until they complete:

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()
engine = create_engine("sqlite:///test.db")
Session = sessionmaker(bind=engine)

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)

# Sync usage - blocks the thread
with Session() as session:
    user = session.query(User).filter_by(name="Alice").first()
    print(user.id)

Asynchronous SQLAlchemy (AsyncIO)

Uses async/await, so your code doesn't block the event loop:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
import asyncio

Base = declarative_base()
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_user():
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            "SELECT * FROM users WHERE name=:name", {"name": "Alice"}
        )
        user = result.fetchone()
        print(user)

asyncio.run(get_user())

Key Differences

Aspect	Synchronous	Asynchronous
Execution	Blocks thread until query finishes	Non-blocking, allows other coroutines to run
Driver	psycopg2, pymysql, sqlite3	asyncpg, aiosqlite, aiomysql
Session	Session	AsyncSession
Query	session.query().all()	await session.execute()
Engine	create_engine()	create_async_engine()
Performance	Good for single-threaded apps	Better for high-concurrency async apps

Important Notes

1. Async is not faster for a single query - It helps when doing many concurrent DB operations
2. Don't mix sync and async sessions - Stick to one pattern in a given context
3. Result handling is different - Use .fetchone(), .fetchall(), or .scalars()

---

Part 3: Dependency Injection

What is Dependency Injection?

Dependency Injection (DI) is a design pattern where functions or objects declare what they need, which are then "injected" into them rather than creating those dependencies themselves.

The Depends() Function

The core mechanism for DI in FastAPI:

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

app = FastAPI()

# Define a dependency
async def get_db_session():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

# Use the dependency
@app.get("/users/")
async def list_users(db: AsyncSession = Depends(get_db_session)):
    result = await db.execute("SELECT * FROM users")
    return result.fetchall()

Database Session Dependency with Yield

Using a generator (yield) that opens a session before the request and closes it afterward:

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_async_db_session():
    """Yields a database session with error handling."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
        except Exception as e:
            await session.rollback()
            raise RuntimeError(f"Database session error: {e!r}") from e

Dependency Injection with Type Hints

Using Annotated for cleaner syntax:

from typing import Annotated
from fastapi import Depends

# Type alias for the dependency
DB_Session = Annotated[AsyncSession, Depends(get_db_session)]

@app.get("/users/")
async def list_users(session: DB_Session):
    result = await session.execute(select(User))
    return result.scalars().all()

Class-Based Dependencies

For stateful services (e.g., cache or email service):

class UserService:
    def __init__(self, db: AsyncSession = Depends(get_db_session)):
        self.db = db

    async def get_user(self, user_id: int):
        result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar()

@app.get("/users/{user_id}")
async def get_user(user_id: int, service: UserService = Depends()):
    return await service.get_user(user_id)

Testing with Dependency Overrides

def get_test_db():
    # Return a test database session
    yield test_session

# Override the dependency for testing
app.dependency_overrides[get_db_session] = get_test_db

---

Part 4: CRUD Operations (Repository Pattern)

Repository Pattern Structure

Organize CRUD functions in a repository/service layer:

project/
├── db.py              # Engine/session configuration
├── deps.py            # get_db_session dependency
├── repositories/
│   ├── __init__.py
│   └── user_repository.py
├── routers/
│   ├── __init__.py
│   └── users.py
└── models/
    └── user.py

User Repository Example

# repositories/user_repository.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models.user import User

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def create(self, name: str, email: str) -> User:
        user = User(name=name, email=email)
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

    async def get_by_id(self, user_id: int) -> User | None:
        result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar()

    async def get_all(self) -> list[User]:
        result = await self.db.execute(select(User))
        return result.scalars().all()

    async def update(self, user_id: int, **kwargs) -> User | None:
        user = await self.get_by_id(user_id)
        if user:
            for key, value in kwargs.items():
                setattr(user, key, value)
            await self.db.commit()
            await self.db.refresh(user)
        return user

    async def delete(self, user_id: int) -> bool:
        user = await self.get_by_id(user_id)
        if user:
            await self.db.delete(user)
            await self.db.commit()
            return True
        return False

Router with Repository

# routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from typing import Annotated
from repositories.user_repository import UserRepository
from deps import get_db_session

router = APIRouter(prefix="/users", tags=["users"])

def get_user_repo(db = Depends(get_db_session)) -> UserRepository:
    return UserRepository(db)

UserRepo = Annotated[UserRepository, Depends(get_user_repo)]

@router.post("/")
async def create_user(name: str, email: str, repo: UserRepo):
    return await repo.create(name, email)

@router.get("/{user_id}")
async def get_user(user_id: int, repo: UserRepo):
    user = await repo.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@router.get("/")
async def list_users(repo: UserRepo):
    return await repo.get_all()

@router.put("/{user_id}")
async def update_user(user_id: int, name: str, repo: UserRepo):
    user = await repo.update(user_id, name=name)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@router.delete("/{user_id}")
async def delete_user(user_id: int, repo: UserRepo):
    if not await repo.delete(user_id):
        raise HTTPException(status_code=404, detail="User not found")
    return {"status": "deleted"}

---

Part 5: ASGI Applications

What is ASGI?

ASGI (Asynchronous Server Gateway Interface) is the successor to WSGI:

• WSGI: For synchronous Python web apps (Flask, Django)
• ASGI: For asynchronous communication (WebSockets, HTTP/2, background tasks)

Concept	Description
ASGI	Async Server Gateway Interface (modern, async version of WSGI)
ASGIApplication	A callable object conforming to ASGI spec

ASGI Application Structure

An ASGI application is any callable that follows the specification:

async def app(scope, receive, send):
    ...

• scope: Information about the connection (type = "http", "websocket", etc.)
• receive: An awaitable that yields incoming events
• send: A callable to send responses back to the client

Example: Minimal ASGI Application

class SimpleASGIApp:
    async def __call__(self, scope, receive, send):
        if scope['type'] == 'http':
            await send({
                'type': 'http.response.start',
                'status': 200,
                'headers': [(b'content-type', b'text/plain')],
            })
            await send({
                'type': 'http.response.body',
                'body': b'Hello, ASGI world!',
            })

# Run with: uvicorn myapp:SimpleASGIApp

ASGI in FastAPI

Key Components:

1. Uvicorn Server: High-performance ASGI server handling network communication
2. FastAPI app: The Python application framework defining API logic
3. H11 Protocol: HTTP/1.1 protocol implementation
4. Router/Routes: Maps incoming request URL and HTTP method to functions

Request Flow:

1. Incoming request arrives at Uvicorn server
2. H11 protocol processes raw HTTP request data
3. ASGI interface calls the application's __call__ method
4. FastAPI's router matches to a defined route
5. Path operation function executes and generates response

---

Part 6: Routes and Dependant

Routing in Starlette

The process starts with a top-level app (ASGI application) that registers Routes:

How DI Works with Routes

1. Request matches a route
2. System inspects the endpoint function signature
3. Identifies required inputs and dependencies
4. Resolves dependencies defined by Depends()
5. Passes resolved dependencies to the endpoint function

@app.get("/users/")
def read_users(p1: str, db: Session = Depends(get_db)):
    # p1 comes from query parameter
    # db is resolved from get_db dependency
    return db.query(User).all()

---

Appendix: Selectors and Event-Driven I/O

The Selectors Module

The selectors module provides a high-level, cross-platform I/O multiplexing interface:

• select() - Basic, cross-platform
• poll() - More efficient on Unix
• epoll() - Linux-specific, highly efficient
• kqueue() - macOS/BSD, highly efficient

Event-Driven Chat Server Example

Using selectors module (cleaner version):

import selectors
import socket

HOST = "127.0.0.1"
PORT = 12345

# Create selector
sel = selectors.DefaultSelector()

def accept(sock):
    client_sock, addr = sock.accept()
    print("Connected from:", addr)
    client_sock.setblocking(False)
    sel.register(client_sock, selectors.EVENT_READ, read_client)

def read_client(client_sock):
    try:
        data = client_sock.recv(4096)
    except BlockingIOError:
        return

    if data:
        msg = data.decode().strip()
        print(f"[{client_sock.fileno()}] {msg}")

        # Broadcast to all clients
        for key in sel.get_map().values():
            s = key.fileobj
            if s not in (server_sock, client_sock):
                try:
                    s.sendall(data)
                except BlockingIOError:
                    pass
    else:
        # Client disconnected
        print(f"Client {client_sock.fileno()} disconnected")
        sel.unregister(client_sock)
        client_sock.close()

# Create server socket
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_sock.bind((HOST, PORT))
server_sock.listen()
server_sock.setblocking(False)

sel.register(server_sock, selectors.EVENT_READ, accept)

print(f"Chat server running on {HOST}:{PORT}")

# Event loop
while True:
    events = sel.select(timeout=1)
    for key, mask in events:
        callback = key.data
        sock = key.fileobj
        callback(sock)

How it works:

1. Server socket: Non-blocking, registered for READ events
2. New client: Accept connection, set non-blocking, register with selector
3. Event loop: Call select() to get ready events
4. Client disconnect: Unregister and close socket

---

📖 References

Resource	Link	Duration
FastAPI Dependencies Tutorial	fastapi.tiangolo.com	~20 min
Dependencies with Yield	fastapi.tiangolo.com	~15 min
SQL Databases Guide	fastapi.tiangolo.com	~25 min
Async SQLAlchemy Sessions	dev.to	~15 min

Learning Checklist:

• Explain what DI is in FastAPI and how Depends() injects reusable logic
• Implement a basic dependency function and inject it via Depends()
• Create a get_db_session dependency using a generator (yield)
• Wire up AsyncEngine + AsyncSession for async stacks
• Integrate DB session dependency into endpoints
• Organize CRUD functions in a repository/service layer
• Test with dependency overrides (app.dependency_overrides)
• Use class-based dependencies for stateful services
• Confirm error handling & teardown occur correctly with exceptions
• Review project structure for DI: db.py, deps.py, repositories/, routers/

---

Summary

This document covers:

1. Dependency Injection: Using Depends() to inject database sessions and services
2. CRUD Operations: Implementing Repository pattern for clean separation
3. ASGI: Understanding how FastAPI works as an ASGI application
4. Selectors: Low-level event-driven I/O for understanding async concepts