Working with SQL Databases

import sqlite3
from contextlib import closing

# Connect and create table
with closing(sqlite3.connect("app.db")) as conn:
    conn.row_factory = sqlite3.Row  # Access columns by name
    
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            email TEXT UNIQUE NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    # Insert with parameter substitution (NEVER use f-strings — SQL injection risk!)
    cursor.execute(
        "INSERT INTO users (username, email) VALUES (?, ?)",
        ("alice", "alice@example.com")
    )
    
    # Insert many
    users = [("bob", "bob@ex.com"), ("carol", "carol@ex.com")]
    cursor.executemany("INSERT INTO users (username, email) VALUES (?, ?)", users)
    
    conn.commit()

# Query
with closing(sqlite3.connect("app.db")) as conn:
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    
    cursor.execute("SELECT * FROM users WHERE id > ?", (0,))
    for row in cursor.fetchall():
        print(f"{row['id']}: {row['username']} <{row['email']}>")

# pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import DeclarativeBase, relationship, Session
from sqlalchemy.sql import func
from datetime import datetime

# Engine — connection to the database
engine = create_engine(
    "sqlite:///app.db",           # SQLite
    # "postgresql://user:pass@localhost/dbname",  # PostgreSQL
    echo=False  # Set True to log all SQL
)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(255), unique=True, nullable=False)
    created_at = Column(DateTime, server_default=func.now())
    
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")
    
    def __repr__(self):
        return f"User(id={self.id}, username={self.username!r})"

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    created_at = Column(DateTime, server_default=func.now())
    
    author = relationship("User", back_populates="posts")

# Create all tables
Base.metadata.create_all(engine)

from sqlalchemy import select, update, delete

# CREATE
with Session(engine) as session:
    user = User(username="alice", email="alice@example.com")
    session.add(user)
    
    # Add multiple
    session.add_all([
        User(username="bob", email="bob@example.com"),
        User(username="carol", email="carol@example.com")
    ])
    
    session.commit()
    print(f"Created user ID: {user.id}")  # ID available after commit

# READ
with Session(engine) as session:
    # Get by primary key
    user = session.get(User, 1)
    
    # Query with filtering
    stmt = select(User).where(User.username == "alice")
    alice = session.scalar(stmt)  # Returns single result or None
    
    # All users
    all_users = session.scalars(select(User)).all()
    
    # With conditions
    stmt = select(User).where(User.id > 1).order_by(User.username).limit(10)
    users = session.scalars(stmt).all()
    
    # Joins
    stmt = (
        select(Post)
        .join(User)
        .where(User.username == "alice")
        .order_by(Post.created_at.desc())
    )
    alice_posts = session.scalars(stmt).all()

# UPDATE
with Session(engine) as session:
    user = session.get(User, 1)
    user.email = "new_email@example.com"
    session.commit()
    
    # Bulk update
    stmt = update(User).where(User.id > 5).values(username="updated")
    session.execute(stmt)
    session.commit()

# DELETE
with Session(engine) as session:
    user = session.get(User, 1)
    session.delete(user)
    session.commit()
    
    # Bulk delete
    stmt = delete(User).where(User.created_at < some_date)
    session.execute(stmt)
    session.commit()

from typing import Optional, List

class UserRepository:
    def __init__(self, session: Session):
        self.session = session
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        return self.session.get(User, user_id)
    
    def get_by_username(self, username: str) -> Optional[User]:
        return self.session.scalar(
            select(User).where(User.username == username)
        )
    
    def get_all(self, skip=0, limit=100) -> List[User]:
        return self.session.scalars(
            select(User).offset(skip).limit(limit)
        ).all()
    
    def create(self, username: str, email: str) -> User:
        user = User(username=username, email=email)
        self.session.add(user)
        self.session.flush()  # Get the ID without committing
        return user
    
    def delete(self, user_id: int) -> bool:
        user = self.get_by_id(user_id)
        if not user:
            return False
        self.session.delete(user)
        return True

# Usage
with Session(engine) as session:
    repo = UserRepository(session)
    
    new_user = repo.create("dave", "dave@example.com")
    user = repo.get_by_username("alice")
    
    session.commit()  # Commit once at the end

# pip install psycopg2-binary sqlalchemy
engine = create_engine(
    "postgresql+psycopg2://user:password@localhost:5432/mydb",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True  # Check connection before using
)

import os
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./app.db")
engine = create_engine(DATABASE_URL)