Lesson 05: Database Layer#

"The database is the source of truth. Everything else is just caching."

📚 Table of Contents#

1. Introduction
2. What is database.py?
3. Line-by-Line Walkthrough
4. SQLAlchemy Components
5. Database Engine
6. Session Factory
7. Declarative Base
8. Migrations
9. Dependency Injection
10. Key Takeaways
11. Interview Questions
12. Exercises
13. Next Steps

Introduction#

The database layer is the foundation of any backend application. In VeyraOps AI, database.py is a small but critical file that:

• Configures the database connection
• Creates the session factory
• Provides the ORM base class
• Handles schema migrations
• Implements dependency injection

This lesson dissects every single line of database.py so you understand exactly how SQLAlchemy works.

What is database.py?#

Location: backend/app/database.py
Size: 54 lines
Purpose: Database configuration and session management

from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import create_engine, text
 
DATABASE_URL = "sqlite:///./agentops.db"
 
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
 
def run_startup_migrations():
    """Apply small SQLite schema updates that create_all will not handle."""
    if not DATABASE_URL.startswith("sqlite"):
        return
    
    tool_call_columns = {
        "model_name": "VARCHAR(100)",
        "estimated_tokens": "INTEGER",
        "estimated_cost": "VARCHAR(30)",
    }
    
    with engine.begin() as connection:
        existing_columns = {
            row[1] for row in connection.execute(text("PRAGMA table_info(tool_calls)"))
        }
        
        for column_name, column_type in tool_call_columns.items():
            if column_name not in existing_columns:
                connection.execute(
                    text(f"ALTER TABLE tool_calls ADD COLUMN {column_name} {column_type}")
                )
 
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Line-by-Line Walkthrough#

Imports#

from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import create_engine, text

What each import does:

• sessionmaker: Factory for creating database sessions
• declarative_base: Base class for ORM models
• create_engine: Creates database connection pool
• text: Allows raw SQL queries (for migrations)

Database URL#

DATABASE_URL = "sqlite:///./agentops.db"

Breakdown:

• sqlite:// — Database driver (SQLite)
• /./ — Current directory
• agentops.db — Database filename

For PostgreSQL (future):

DATABASE_URL = "postgresql://user:password@localhost:5432/agentops"

For MySQL:

DATABASE_URL = "mysql://user:password@localhost:3306/agentops"

Why SQLite?

• ✅ No separate server needed
• ✅ Zero configuration
• ✅ Perfect for development
• ✅ File-based (easy backup)

Database Engine#

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}
)

What is an Engine? The engine is the connection pool manager. It:

• Maintains database connections
• Reuses connections (efficient)
• Handles connection lifecycle

connect_args={"check_same_thread": False}

Why needed? SQLite by default only allows connections in the thread that created them. FastAPI is multi-threaded, so we disable this check.

⚠️ Safety: This is safe because SQLAlchemy handles thread safety internally.

PostgreSQL doesn't need this:

engine = create_engine("postgresql://...")  # No connect_args needed

Session Factory#

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

What is a Session? A session is like a "shopping cart" for database operations:

• Add items (models) to it
• Commit when ready
• Rollback if errors occur

Parameters explained:

autocommit=False

• Changes only saved when you call db.commit()
• Gives you control over transactions
• Can rollback if errors occur

autoflush=False

• SQLAlchemy won't automatically execute SQL
• You control when queries run
• More predictable behavior

bind=engine

• Links this session factory to the database engine
• All sessions from this factory use the same database

Declarative Base#

Base = declarative_base()

What is Base? The parent class for all ORM models.

Usage:

# In models.py
class Agent(Base):
    __tablename__ = "agents"
    # ...

What Base provides:

• ✅ Metadata tracking (all tables)
• ✅ Table creation capability
• ✅ Relationship management
• ✅ Query interface

Creating tables:

Base.metadata.create_all(bind=engine)

This scans all classes inheriting from Base and creates their tables.

Migrations#

def run_startup_migrations():
    """Apply small SQLite schema updates that create_all will not handle."""
    if not DATABASE_URL.startswith("sqlite"):
        return

What are migrations? Changes to database schema after tables are created.

Why needed? create_all() only creates tables that don't exist. It won't add new columns to existing tables.

Example problem:

# Version 1: models.py
class ToolCall(Base):
    id = Column(Integer)
    tool_name = Column(String)
    # Database created with these 2 columns
 
# Version 2: models.py (later)
class ToolCall(Base):
    id = Column(Integer)
    tool_name = Column(String)
    model_name = Column(String)  # ← NEW COLUMN
 
# Problem: create_all() won't add model_name to existing table!

Solution: Manual migration

Migration Logic#

tool_call_columns = {
    "model_name": "VARCHAR(100)",
    "estimated_tokens": "INTEGER",
    "estimated_cost": "VARCHAR(30)",
}

Dictionary of new columns:

• Key: Column name
• Value: SQL type

Check Existing Columns#

with engine.begin() as connection:
    existing_columns = {
        row[1] for row in connection.execute(text("PRAGMA table_info(tool_calls)"))
    }

engine.begin(): Starts a transaction
connection.execute(): Runs raw SQL
text("PRAGMA table_info(tool_calls)"): SQLite command to list columns

Output:

(0, 'id', 'INTEGER', ...)
(1, 'tool_name', 'VARCHAR', ...)
(2, 'model_name', 'VARCHAR', ...)  # If exists

row[1]: Extract column name (second element)

Result: Set of existing column names

{'id', 'tool_name', 'model_name'}

Add Missing Columns#

for column_name, column_type in tool_call_columns.items():
    if column_name not in existing_columns:
        connection.execute(
            text(f"ALTER TABLE tool_calls ADD COLUMN {column_name} {column_type}")
        )

Logic:

1. Loop through desired columns
2. Check if column exists
3. If not, add it with ALTER TABLE

SQL generated:

ALTER TABLE tool_calls ADD COLUMN model_name VARCHAR(100);
ALTER TABLE tool_calls ADD COLUMN estimated_tokens INTEGER;
ALTER TABLE tool_calls ADD COLUMN estimated_cost VARCHAR(30);

Why f-string is safe here:

• We control the values (not user input)
• No SQL injection risk

For user input, use parameters:

# BAD (SQL injection risk)
connection.execute(text(f"SELECT * FROM users WHERE name = '{user_input}'"))
 
# GOOD (parameterized)
connection.execute(text("SELECT * FROM users WHERE name = :name"), {"name": user_input})

Dependency Injection#

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

What is this pattern? A generator function that FastAPI uses for dependency injection.

Execution flow:

1. Request arrives
2. db = SessionLocal() — Create new session
3. yield db — Provide session to endpoint
4. Endpoint executes — Uses db parameter
5. finally: db.close() — Always close session, even if error

Usage in routers:

@router.post("/agents/{agent_id}/chat")
def chat(agent_id: int, db: Session = Depends(get_db)):
    agent = db.query(Agent).filter(Agent.id == agent_id).first()
    # ...

Why yield instead of return?

# With return - NO CLEANUP
def get_db():
    db = SessionLocal()
    return db  # ❌ db never closes
 
# With yield - AUTOMATIC CLEANUP  
def get_db():
    db = SessionLocal()
    try:
        yield db  # ✅ Pauses here, endpoint runs
    finally:
        db.close()  # ✅ Always executes after endpoint

SQLAlchemy Components#

Engine vs Session vs Connection#

┌─────────────────────────────────────┐
│           Application               │
└─────────────────────────────────────┘
                  ↓
┌─────────────────────────────────────┐
│         Session (ORM)               │
│  - Python objects                   │
│  - Tracks changes                   │
│  - commit() / rollback()            │
└─────────────────────────────────────┘
                  ↓
┌─────────────────────────────────────┐
│         Engine (Connection Pool)    │
│  - Manages connections              │
│  - Reuses connections               │
│  - Thread-safe                      │
└─────────────────────────────────────┘
                  ↓
┌─────────────────────────────────────┐
│         Database (SQLite)           │
│  - Stores data on disk              │
└─────────────────────────────────────┘

Transaction Lifecycle#

# 1. Create session
db = SessionLocal()
 
# 2. Query database (SELECT)
agent = db.query(Agent).filter(Agent.id == 1).first()
 
# 3. Modify object
agent.name = "New Name"
 
# 4. Add new object
new_message = Message(content="Hello")
db.add(new_message)
 
# 5. Commit transaction (saves to database)
db.commit()
 
# 6. Close session
db.close()

With error handling:

db = SessionLocal()
try:
    agent = db.query(Agent).first()
    agent.name = "New Name"
    db.commit()
except Exception as e:
    db.rollback()  # Undo changes
    raise
finally:
    db.close()  # Always close

Key Takeaways#

🎯 database.py configures the database engine, session factory, and ORM base.

🎯 The engine manages the connection pool and is created once per application.

🎯 SessionLocal is a factory that creates new sessions per request.

🎯 Base is the parent class for all ORM models, enabling table creation.

🎯 Migrations handle schema changes that create_all() cannot.

🎯 get_db() provides dependency injection with automatic cleanup.

🎯 yield ensures sessions are always closed, even if errors occur.

Interview Questions#

Junior Level#

Q1: What is the purpose of database.py?
A: Configures the database connection, creates the session factory, provides the ORM base class, and implements dependency injection.

Q2: What does SessionLocal() do?
A: Creates a new database session for performing queries and transactions.

Q3: Why do we use yield in get_db()?
A: To ensure the session is always closed after use, even if an error occurs.

Mid-Level#

Q4: Explain the difference between the engine and a session.
A: The engine manages the connection pool (shared across the app). A session represents a single unit of work (one request) and tracks changes to objects.

Q5: Why is autocommit=False important?
A: It gives us transaction control. Changes are only saved when we explicitly call commit(), and we can rollback() if errors occur.

Q6: What problem do the startup migrations solve?
A: create_all() only creates tables that don't exist. It won't add new columns to existing tables. Migrations handle these schema changes.

Senior Level#

Q7: How would you implement connection pooling for PostgreSQL in production?
A: Use create_engine() with pool_size and max_overflow:

engine = create_engine(
    DATABASE_URL,
    pool_size=20,          # Base connections
    max_overflow=10,       # Extra connections under load
    pool_pre_ping=True,    # Check connection health
    pool_recycle=3600      # Recycle connections after 1 hour
)

Q8: The database is experiencing deadlocks. How would you debug and fix it?
A: (1) Enable SQLAlchemy logging to see queries. (2) Check for long-running transactions. (3) Ensure consistent lock ordering. (4) Add isolation_level="READ COMMITTED" to engine. (5) Use with_for_update() for explicit row locking.

Q9: Design a migration system using Alembic instead of manual migrations.
A: Install Alembic, create alembic.ini and alembic/ folder. Generate migrations with alembic revision --autogenerate. Alembic tracks version history and can upgrade/downgrade. Replace run_startup_migrations() with alembic upgrade head in deployment.

Exercises#

Exercise 1: Add Logging#

Task: Add SQLAlchemy logging to see all SQL queries executed.

import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

Exercise 2: Connection Pool Test#

Task: Create 100 concurrent requests and verify the connection pool is reusing connections.

Exercise 3: Migration Practice#

Task: Add a new column priority to the agents table using the migration pattern.

Exercise 4: Transaction Rollback#

Task: Write code that demonstrates a rollback when an error occurs mid-transaction.

Exercise 5: PostgreSQL Migration#

Task: Modify database.py to support PostgreSQL using an environment variable.

Next Steps#

You now understand database.py and how SQLAlchemy manages connections.

Next, you'll learn about models.py — every table, column, and relationship in the system.

👉 Lesson 06: Database Models →

← Previous: Folder Structure | ↑ Back to Index | Next: Models →

Module 1 · Foundation