Database Tutorial

This tutorial covers using Tet’s enhanced SQLAlchemy integration, including root factories, session management, and database best practices.

Database Setup

Setting up SQLAlchemy with Tet’s enhancements.

Basic Configuration

Create your database models and configuration:

# models.py
from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from datetime import datetime, timezone

Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    password_hash = Column(String(128))
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    is_active = Column(Boolean, default=True)

    # Relationships
    posts = relationship("Post", back_populates="author")

    def __json__(self, request):
        """JSON serialization for Tet's renderer."""
        return {
            "id": self.id,
            "username": self.username,
            "email": self.email,
            "created_at": self.created_at,
            "is_active": self.is_active,
        }


class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    author_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    is_published = Column(Boolean, default=False)

    # Relationships
    author = relationship("User", back_populates="posts")

    def __json__(self, request):
        return {
            "id": self.id,
            "title": self.title,
            "content": self.content,
            "author_id": self.author_id,
            "created_at": self.created_at,
            "is_published": self.is_published,
        }

Database Engine Setup

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base


def get_engine(settings):
    """Create database engine from settings."""
    return create_engine(
        settings["sqlalchemy.url"],
        echo=settings.get("sqlalchemy.echo", False),
        pool_size=int(settings.get("sqlalchemy.pool_size", 10)),
        max_overflow=int(settings.get("sqlalchemy.max_overflow", 20)),
        pool_timeout=int(settings.get("sqlalchemy.pool_timeout", 30)),
        pool_recycle=int(settings.get("sqlalchemy.pool_recycle", 3600)),
    )


def get_session_factory(engine):
    """Create session factory."""
    return sessionmaker(bind=engine)


def initialize_database(engine):
    """Initialize database tables."""
    Base.metadata.create_all(engine)

Request-Scoped Sessions with Tet

Tet provides automatic database session management through pyramid_di and pyramid_tm.

Tet’s SQLAlchemy Integration

# app.py
from tet.config import application_factory, ALL_FEATURES
from tet.sqlalchemy.simple import declarative_base

# Create declarative base using Tet's helper
Base = declarative_base()


@application_factory(included_features=ALL_FEATURES)
def main(config):
    """Tet application factory with database support."""

    # Include Tet's simple SQLAlchemy setup
    config.include("tet.sqlalchemy.simple")

    # Setup SQLAlchemy - automatically configures:
    # - pyramid_di service registration
    # - pyramid_tm transaction management
    # - Request-scoped sessions
    config.setup_sqlalchemy()

    # Routes and views
    config.add_route("users", "/users")
    config.add_route("user", "/users/{id}")
    config.scan("views")

Settings Configuration

Configure your database in settings (e.g., development.ini):

[app:main]
use = egg:myapp

# SQLAlchemy configuration
sqlalchemy.url = sqlite:///myapp.db
sqlalchemy.echo = false
sqlalchemy.pool_size = 10

Automatic Features Included

When you use config.setup_sqlalchemy(), Tet automatically configures:

1. pyramid_di: For dependency injection

2. pyramid_tm: For transaction management

3. Request-scoped sessions: Sessions tied to request lifecycle

4. Automatic commit/rollback: Based on HTTP response status

5. Session cleanup: Sessions closed automatically after request

Root Factories

Use Tet’s SQLARootFactory for traversal-based applications.

Basic Root Factory

# root.py
from tet.sqlalchemy.factory import SQLARootFactory
from models import User, Post
from sqlalchemy.orm.exc import NoResultFound


class UserRootFactory(SQLARootFactory):
    """Root factory for user resources."""

    def supplier(self, item):
        """Look up user by ID."""
        session = self.request.find_service(name="dbsession")
        try:
            return session.query(User).filter_by(id=int(item)).one()
        except (NoResultFound, ValueError):
            # These exceptions are converted to KeyError (404)
            raise


class PostRootFactory(SQLARootFactory):
    """Root factory for post resources."""

    def supplier(self, item):
        """Look up post by ID or slug."""
        session = self.request.find_service(name="dbsession")

        try:
            # Try numeric ID first
            post_id = int(item)
            return session.query(Post).filter_by(id=post_id).one()
        except ValueError:
            # Try as slug
            try:
                return session.query(Post).filter_by(slug=item).one()
            except NoResultFound:
                raise

Multi-Model Root Factory

class ApplicationRootFactory(SQLARootFactory):
    """Root factory that handles multiple resource types."""

    def supplier(self, item):
        """Route to appropriate model based on item format."""
        session = self.request.find_service(name="dbsession")

        # Handle different patterns
        if item.startswith("user-"):
            user_id = item[5:]  # Remove 'user-' prefix
            try:
                return session.query(User).filter_by(id=int(user_id)).one()
            except (NoResultFound, ValueError):
                raise

        elif item.startswith("post-"):
            post_id = item[5:]  # Remove 'post-' prefix
            try:
                return session.query(Post).filter_by(id=int(post_id)).one()
            except (NoResultFound, ValueError):
                raise

        else:
            # Try as numeric ID for backwards compatibility
            try:
                # Default to User lookup
                return session.query(User).filter_by(id=int(item)).one()
            except (NoResultFound, ValueError):
                raise

Using Root Factories

Configure traversal with root factories:

def main(global_config, **settings):
    with Configurator(settings=settings) as config:
        # Set root factory for traversal
        config.set_root_factory(ApplicationRootFactory)

        # Add traversal routes
        config.add_route("user_detail", "/users/*traverse")
        config.add_route("post_detail", "/posts/*traverse")

        return config.make_wsgi_app()

Database Views

Create views that work with Tet’s database integration.

User Management Views

# views/users.py
from pyramid.view import view_config
from pyramid.httpexceptions import HTTPNotFound, HTTPBadRequest
from pyramid_di import autowired
from sqlalchemy.orm import Session
from models import User


class UserViews:
    """User management views with autowired dependencies."""

    # Database session automatically injected via pyramid_di
    session: Session = autowired()

    @view_config(route_name="users", request_method="GET", renderer="json")
    def list_users(self, request):
        """List all users."""
        # Pagination
        page = int(request.params.get("page", 1))
        per_page = int(request.params.get("per_page", 20))

        query = self.session.query(User).filter_by(is_active=True)

        # Apply filters
        if "search" in request.params:
            search = f"%{request.params['search']}%"
            query = query.filter(User.username.ilike(search))

        # Paginate
        total = query.count()
        users = query.offset((page - 1) * per_page).limit(per_page).all()

        return {
            "users": users,  # Automatically serialized by Tet
            "pagination": {
                "page": page,
                "per_page": per_page,
                "total": total,
                "pages": (total + per_page - 1) // per_page,
            },
        }

    @view_config(route_name="user", request_method="GET", renderer="json")
    def get_user(self, request):
        """Get single user by ID."""
        user_id = request.matchdict["id"]

        try:
            user = self.session.query(User).filter_by(id=int(user_id)).one()
            return {"user": user}
        except (NoResultFound, ValueError):
            raise HTTPNotFound("User not found")

    @view_config(route_name="users", request_method="POST", renderer="json")
    def create_user(self, request):
        """Create new user."""
        data = request.json_body

        # Validation
        if not data.get("username") or not data.get("email"):
            raise HTTPBadRequest("Username and email are required")

        # Check for existing user
        existing = (
            self.session.query(User)
            .filter((User.username == data["username"]) | (User.email == data["email"]))
            .first()
        )

        if existing:
            raise HTTPBadRequest("Username or email already exists")

        # Create user
        user = User(
            username=data["username"],
            email=data["email"],
            password_hash=hash_password(data.get("password", "")),
        )

        self.session.add(user)
        # No manual commit needed - pyramid_tm handles it automatically

        return {"user": user, "message": "User created successfully"}

Relationship Handling

@view_config(route_name="user_posts", renderer="json")
def get_user_posts(request):
    """Get posts by user."""
    user_id = request.matchdict["user_id"]
    session = request.find_service(name="dbsession")

    try:
        user = session.query(User).filter_by(id=int(user_id)).one()
    except (NoResultFound, ValueError):
        raise HTTPNotFound("User not found")

    # Get user's posts with eager loading
    posts = (
        session.query(Post)
        .filter_by(author_id=user.id, is_published=True)
        .order_by(Post.created_at.desc())
        .all()
    )

    return {"user": user, "posts": posts, "post_count": len(posts)}

Complex Queries

from sqlalchemy import func, desc


@view_config(route_name="user_stats", renderer="json")
def user_statistics(request):
    """Get user statistics."""
    session = request.find_service(name="dbsession")

    # Complex query with aggregations
    stats = (
        session.query(
            User.id,
            User.username,
            func.count(Post.id).label("post_count"),
            func.max(Post.created_at).label("last_post_date"),
        )
        .outerjoin(Post)
        .group_by(User.id)
        .all()
    )

    # The results are automatically JSON-serializable
    return {"user_stats": stats}


@view_config(route_name="popular_posts", renderer="json")
def popular_posts(request):
    """Get popular posts with author info."""
    session = request.find_service(name="dbsession")

    # Join query with eager loading
    posts = (
        session.query(Post)
        .join(User)
        .filter(Post.is_published == True)
        .order_by(desc(Post.created_at))
        .limit(10)
        .all()
    )

    # Custom serialization including author info
    result = []
    for post in posts:
        result.append(
            {
                "id": post.id,
                "title": post.title,
                "content": post.content[:200] + "...",  # Truncate
                "created_at": post.created_at,
                "author": {"id": post.author.id, "username": post.author.username},
            }
        )

    return {"posts": result}

Transaction Management

Tet automatically handles database transactions through pyramid_tm.

Automatic Transaction Management

With Tet’s SQLAlchemy setup, transactions are handled automatically:

@application_factory(included_features=ALL_FEATURES)
def main(config):
    """Tet automatically includes pyramid_tm."""
    config.include("tet.sqlalchemy.simple")
    config.setup_sqlalchemy()

    # pyramid_tm is automatically included and configured
    # Transactions are:
    # - Started for each request
    # - Committed on successful response (2xx, 3xx)
    # - Rolled back on exceptions or error responses (4xx, 5xx)

Working with Automatic Transactions

Your views don’t need to manage transactions manually:

@view_config(route_name="complex_operation", renderer="json")
def complex_database_operation(request):
    """Complex operation with automatic transaction management."""
    session = request.find_service(Session)

    # All operations happen in one transaction
    # Create user
    user = User(username="newuser", email="new@example.com")
    session.add(user)
    session.flush()  # Get the ID without committing

    # Create initial post
    post = Post(
        title="Welcome Post",
        content="Welcome to the platform!",
        author_id=user.id,
        is_published=True,
    )
    session.add(post)

    # If view returns successfully, transaction commits automatically
    # If any exception is raised, transaction rolls back automatically
    return {"message": "Operation completed successfully", "user": user}

Handling Transaction Rollbacks

To trigger a rollback, raise an HTTP exception:

from pyramid.httpexceptions import HTTPBadRequest


@view_config(route_name="conditional_operation", renderer="json")
def conditional_operation(request):
    session = request.find_service(Session)

    # Do some work
    user = User(username="test")
    session.add(user)

    # Check some condition
    if some_validation_fails():
        # This will cause automatic transaction rollback
        raise HTTPBadRequest("Validation failed")

    # If we reach here, transaction will commit automatically
    return {"user": user}

Bulk Operations

@view_config(route_name="bulk_import", request_method="POST", renderer="json")
def bulk_import_users(request):
    """Bulk import users efficiently."""
    session = request.find_service(name="dbsession")
    users_data = request.json_body.get("users", [])

    try:
        # Bulk insert for better performance
        user_objects = []
        for user_data in users_data:
            user = User(
                username=user_data["username"],
                email=user_data["email"],
                password_hash=hash_password(user_data.get("password", "")),
            )
            user_objects.append(user)

        # Bulk save
        session.bulk_save_objects(user_objects)
        session.commit()

        return {
            "message": f"Successfully imported {len(user_objects)} users",
            "count": len(user_objects),
        }

    except Exception as e:
        session.rollback()
        raise HTTPBadRequest(f"Bulk import failed: {str(e)}")

Database Migrations

Handle database schema changes.

Simple Migration System

# migrations.py
from sqlalchemy import text

MIGRATIONS = {
    "001_add_user_bio": """
        ALTER TABLE users ADD COLUMN bio TEXT;
    """,
    "002_add_post_slug": """
        ALTER TABLE posts ADD COLUMN slug VARCHAR(200);
        CREATE INDEX idx_posts_slug ON posts(slug);
    """,
    "003_add_timestamps": """
        ALTER TABLE users ADD COLUMN updated_at TIMESTAMP;
        ALTER TABLE posts ADD COLUMN updated_at TIMESTAMP;
    """,
}


def run_migrations(engine):
    """Run pending migrations."""
    with engine.connect() as conn:
        # Create migrations table if it doesn't exist
        conn.execute(
            text("""
            CREATE TABLE IF NOT EXISTS migrations (
                id VARCHAR(255) PRIMARY KEY,
                applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        )

        # Get applied migrations
        result = conn.execute(text("SELECT id FROM migrations"))
        applied = {row[0] for row in result}

        # Run pending migrations
        for migration_id, sql in MIGRATIONS.items():
            if migration_id not in applied:
                print(f"Running migration: {migration_id}")
                conn.execute(text(sql))
                conn.execute(
                    text("INSERT INTO migrations (id) VALUES (:id)"),
                    {"id": migration_id},
                )
                conn.commit()

Using Alembic

For production applications, use Alembic for migrations:

pip install alembic
alembic init alembic
alembic revision --autogenerate -m "Initial migration"
alembic upgrade head

Testing Database Code

Test your database interactions thoroughly.

Database Test Setup

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, User, Post


@pytest.fixture(scope="session")
def engine():
    """Create test database engine."""
    return create_engine("sqlite:///:memory:", echo=False)


@pytest.fixture(scope="session")
def tables(engine):
    """Create all tables for testing."""
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)


@pytest.fixture(scope="function")
def dbsession(engine, tables):
    """Create database session for each test."""
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.rollback()
    session.close()


@pytest.fixture
def sample_user(dbsession):
    """Create sample user for testing."""
    user = User(
        username="testuser", email="test@example.com", password_hash="hashed_password"
    )
    dbsession.add(user)
    dbsession.commit()
    return user

Testing Views with Database

# test_views.py
def test_create_user(dbsession):
    from views.users import create_user
    from pyramid.testing import DummyRequest

    # Mock request
    request = DummyRequest()
    request.find_service = lambda name: dbsession
    request.json_body = {
        "username": "newuser",
        "email": "new@example.com",
        "password": "password123",
    }

    # Test the view
    result = create_user(request)

    assert result["message"] == "User created successfully"
    assert result["user"].username == "newuser"

    # Verify in database
    user = dbsession.query(User).filter_by(username="newuser").first()
    assert user is not None
    assert user.email == "new@example.com"


def test_user_posts(dbsession, sample_user):
    from views.users import get_user_posts
    from pyramid.testing import DummyRequest

    # Create test posts
    post1 = Post(title="Post 1", author_id=sample_user.id, is_published=True)
    post2 = Post(title="Post 2", author_id=sample_user.id, is_published=False)
    dbsession.add_all([post1, post2])
    dbsession.commit()

    # Test the view
    request = DummyRequest()
    request.find_service = lambda name: dbsession
    request.matchdict = {"user_id": str(sample_user.id)}

    result = get_user_posts(request)

    assert result["user"].id == sample_user.id
    assert len(result["posts"]) == 1  # Only published posts
    assert result["posts"][0].title == "Post 1"

Testing Root Factories

def test_user_root_factory(dbsession, sample_user):
    from root import UserRootFactory
    from pyramid.testing import DummyRequest

    request = DummyRequest()
    request.find_service = lambda name: dbsession

    root = UserRootFactory(request)

    # Test successful lookup
    found_user = root[str(sample_user.id)]
    assert found_user == sample_user

    # Test not found
    with pytest.raises(KeyError):
        root["999"]

Performance Optimization

Optimize database performance.

Query Optimization

from sqlalchemy.orm import joinedload, selectinload


@view_config(route_name="optimized_posts", renderer="json")
def optimized_posts_view(request):
    """Optimized post loading with eager loading."""
    session = request.find_service(name="dbsession")

    # Eager load relationships to avoid N+1 queries
    posts = (
        session.query(Post)
        .options(
            joinedload(Post.author),  # Join load for one-to-one/many-to-one
            selectinload(Post.comments),  # Select load for one-to-many
        )
        .filter_by(is_published=True)
        .all()
    )

    return {"posts": posts}

Connection Pooling

def get_engine(settings):
    """Configure engine with optimized connection pooling."""
    return create_engine(
        settings["sqlalchemy.url"],
        # Connection pool settings
        pool_size=20,  # Number of connections to maintain
        max_overflow=50,  # Additional connections beyond pool_size
        pool_timeout=30,  # Seconds to wait for connection
        pool_recycle=3600,  # Recycle connections after 1 hour
        # Query optimization
        echo=False,  # Don't log SQL in production
        echo_pool=False,  # Don't log pool events
    )

Query Caching

from functools import lru_cache


@lru_cache(maxsize=100)
def get_popular_tags():
    """Cache popular tags query."""
    # This would need to be implemented with cache invalidation
    # in a real application
    pass

Best Practices

Session Management - Always use request-scoped sessions - Ensure sessions are properly closed - Use transaction management (pyramid_tm)

Query Optimization - Use eager loading to avoid N+1 queries - Only select needed columns for large datasets - Use proper indexing on frequently queried columns

Error Handling - Use Tet’s root factories for proper exception handling - Handle database exceptions gracefully - Provide meaningful error messages

Security - Always use parameterized queries (SQLAlchemy ORM does this) - Validate input data before database operations - Don’t expose sensitive data in JSON responses

Testing - Use isolated test databases - Test both success and failure scenarios - Use fixtures for consistent test data

Performance - Configure appropriate connection pooling - Use database migrations for schema changes - Monitor query performance in production