Python y bases de datos

import sqlite3
from contextlib import closing

# Connect (creates file if not exists)
conn = sqlite3.connect("app.db")
# conn = sqlite3.connect(":memory:")  # In-memory database

# Row factory for dict-like access
conn.row_factory = sqlite3.Row

# Create tables
conn.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# Insert data (always use parameterized queries!)
conn.execute(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    ("Alice", "alice@example.com", 30)
)

# Insert many
users = [
    ("Bob", "bob@example.com", 25),
    ("Charlie", "charlie@example.com", 35),
    ("Diana", "diana@example.com", 28),
]
conn.executemany(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    users
)
conn.commit()

# Query data
cursor = conn.execute("SELECT * FROM users WHERE age > ?", (27,))
for row in cursor:
    print(f"{row['name']} ({row['email']}): age {row['age']}")

# Fetch methods
cursor = conn.execute("SELECT * FROM users")
first = cursor.fetchone()    # Single row
rest = cursor.fetchall()     # All remaining rows

# Context manager for automatic commit/rollback
with conn:
    conn.execute("UPDATE users SET age = ? WHERE name = ?", (31, "Alice"))

conn.close()

# Best practice: use closing() context manager
with closing(sqlite3.connect("app.db")) as conn:
    conn.row_factory = sqlite3.Row
    with conn:
        rows = conn.execute("SELECT * FROM users").fetchall()

# pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Session, relationship, sessionmaker

# Define the base class
class Base(DeclarativeBase):
    pass

# Define models
class Author(Base):
    __tablename__ = "authors"

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(200), unique=True)

    books = relationship("Book", back_populates="author", cascade="all, delete-orphan")

    def __repr__(self):
        return f"Author(id={self.id}, name='{self.name}')"

class Book(Base):
    __tablename__ = "books"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    price = Column(Float, nullable=False)
    author_id = Column(Integer, ForeignKey("authors.id"), nullable=False)

    author = relationship("Author", back_populates="books")

    def __repr__(self):
        return f"Book(id={self.id}, title='{self.title}')"

# Create engine and tables
engine = create_engine("sqlite:///bookstore.db", echo=False)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)

# CRUD operations
with SessionLocal() as session:
    # Create
    author = Author(name="J.K. Rowling", email="jk@example.com")
    author.books.append(Book(title="Harry Potter 1", price=12.99))
    author.books.append(Book(title="Harry Potter 2", price=14.99))
    session.add(author)
    session.commit()

    # Read
    authors = session.execute(select(Author)).scalars().all()
    for a in authors:
        print(f"{a.name}: {[b.title for b in a.books]}")

    # Query with filters
    expensive = session.execute(
        select(Book).where(Book.price > 13.00)
    ).scalars().all()

    # Join query
    results = session.execute(
        select(Book, Author)
        .join(Author)
        .where(Author.name == "J.K. Rowling")
    ).all()

    # Update
    book = session.execute(
        select(Book).where(Book.title == "Harry Potter 1")
    ).scalar_one()
    book.price = 15.99
    session.commit()

    # Delete
    session.delete(book)
    session.commit()

# pip install sqlalchemy[asyncio] aiosqlite
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select

# Async engine
async_engine = create_async_engine("sqlite+aiosqlite:///async_app.db")
AsyncSessionLocal = async_sessionmaker(async_engine, class_=AsyncSession)

async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User))
        users = result.scalars().all()
        return users

# With FastAPI
from fastapi import Depends

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

# pip install alembic

# Initialize Alembic
# alembic init migrations

# Generate a migration from model changes
# alembic revision --autogenerate -m "add users table"

# Apply migrations
# alembic upgrade head

# Rollback
# alembic downgrade -1

# Example migration file
# def upgrade():
#     op.create_table(
#         'users',
#         sa.Column('id', sa.Integer(), primary_key=True),
#         sa.Column('name', sa.String(100), nullable=False),
#         sa.Column('email', sa.String(200), unique=True),
#     )
#
# def downgrade():
#     op.drop_table('users')