SQLAlchemy and Alembic

1

SQLAlchemy and Alembic are powerful tools for database interactions and schema migrations. Here are some advanced concepts and features you can explore to deepen your expertise:

---

SQLAlchemy Advanced Concepts

1. Advanced ORM Features

• Polymorphic Associations: Use inheritance to handle related tables sharing a common structure. For example:

  class Employee(Base):
      __tablename__ = 'employee'
      id = Column(Integer, primary_key=True)
      type = Column(String)
  
      __mapper_args__ = {
          'polymorphic_on': type,
          'polymorphic_identity': 'employee'
      }
  
  class Manager(Employee):
      __mapper_args__ = {'polymorphic_identity': 'manager'}

• Association Proxies: Simplify relationships with attributes that "proxy" to related objects.

  class User(Base):
      __tablename__ = 'user'
      id = Column(Integer, primary_key=True)
      addresses = relationship('Address')
      email_addresses = association_proxy('addresses', 'email')

2. Custom SQL Expressions

• Use SQLAlchemy’s func for custom SQL functions or write raw SQL within ORM queries.

  from sqlalchemy.sql import func
  session.query(func.count(User.id)).scalar()

3. Hybrid Properties

• Combine Python attributes and SQL expressions.

  from sqlalchemy.ext.hybrid import hybrid_property
  
  class User(Base):
      @hybrid_property
      def full_name(self):
          return f"{self.first_name} {self.last_name}"

4. Event Listeners

• Hook into the SQLAlchemy lifecycle.

  from sqlalchemy.event import listens_for
  
  @listens_for(User, 'before_insert')
  def receive_before_insert(mapper, connection, target):
      target.created_at = datetime.utcnow()

5. Custom Type Definitions

• Define custom column types.

  from sqlalchemy.types import TypeDecorator
  
  class JsonType(TypeDecorator):
      impl = String
  
      def process_bind_param(self, value, dialect):
          return json.dumps(value)
  
      def process_result_value(self, value, dialect):
          return json.loads(value)

6. Declarative Mixins

• Reuse common fields and methods across models.

  class TimestampMixin:
      created_at = Column(DateTime, default=datetime.utcnow)
      updated_at = Column(DateTime, onupdate=datetime.utcnow)
  
  class User(TimestampMixin, Base):
      __tablename__ = 'user'

7. Partitioning and Sharding

• Distribute data across multiple databases or tables with custom session factories and engines.

8. Custom Query Classes

• Extend the Query class for specific query logic.

  class CustomQuery(Query):
      def active(self):
          return self.filter_by(active=True)

---

Alembic Advanced Concepts

1. Environment Customization

• Modify the env.py to dynamically determine configurations, e.g., different databases for production and testing.

2. Branching Migrations

• Handle diverging schema versions using branches.

  alembic revision --branch-label branch_a -m "Add branch A"

3. Custom Migration Operations

• Extend Alembic's Operations for reusable logic.

  from alembic.operations import Operations, MigrateOperation
  
  @Operations.register_operation("create_view")
  def create_view(operations, view_name, selectable):
      operations.execute(f"CREATE VIEW {view_name} AS {selectable}")

4. Autogenerate with Custom Exclusions

• Configure autogeneration rules to exclude certain tables or columns in env.py.

  def include_object(object, name, type_, reflected, compare_to):
      if type_ == "table" and name == "exclude_table":
          return False
      return True

5. Using Templates

• Create custom revision templates for consistent migration scripts.

  alembic revision --template custom_template

6. Data Migrations

• Use migration scripts to modify data, not just schema.

  def upgrade():
      op.execute("UPDATE user SET active = true WHERE last_login > now() - interval '30 days'")

7. Running Alembic Programmatically

• Integrate migrations into Python scripts.

  from alembic import command
  from alembic.config import Config
  
  alembic_cfg = Config("alembic.ini")
  command.upgrade(alembic_cfg, "head")

8. Versioning Multiple Databases

• Manage multiple databases with different migration paths by customizing the env.py.

9. Downgrade and Rollback Strategies

• Implement reversible migrations with robust downgrade scripts.

  def downgrade():
      op.drop_column('user', 'email')

10. Dynamic Migrations

• Use Python code to define migrations dynamically.

  def upgrade():
      metadata = MetaData(bind=op.get_bind())
      user_table = Table('user', metadata, autoload=True)
      op.add_column(user_table, Column('new_column', String))

---

---

2

Here are 10 more complex topics and techniques for SQLAlchemy and Alembic:

---

1. Multi-Tenancy with SQLAlchemy

• Implementing multi-tenancy using schema-per-tenant or database-per-tenant strategies.

• Dynamically setting the schema or database connection for tenants.

---

2. Advanced Query Optimizations

• Using CTEs (Common Table Expressions) for hierarchical or recursive queries.

• Leveraging database-specific features like materialized views through SQLAlchemy.

• Handling query caching with external libraries or in-memory solutions.

---

3. Hybrid Attributes and Custom Queryable Properties

• Creating hybrid attributes that can work in both Python and SQLAlchemy queries.

• Combining them with decorators like @hybrid_property and @hybrid_method.

---

4. Custom Type Decorators

• Defining custom types (e.g., JSONB, encrypted fields) by subclassing TypeDecorator or Mutable.

• Automatically converting Python objects to database-storable types.

---

5. Dynamic Model Creation

• Creating database models dynamically at runtime using Python metaclasses or factory methods.

• Useful for applications requiring flexible data structures (e.g., NoSQL-like behavior).

---

6. Bulk Inserts with Performance Optimization

• Using bulk_save_objects and bulk_insert_mappings for efficient batch inserts.

• Balancing between performance and ORM features (e.g., auto-expiration of session).

---

7. Alembic Custom Operations

• Creating custom operations in Alembic migration scripts, like complex data transformations.

• Writing reusable functions to manage custom database constraints or triggers.

---

8. Database Encryption with SQLAlchemy

• Encrypting sensitive fields at the application level using libraries like Fernet or custom encryption.

• Writing TypeDecorator to ensure seamless encryption/decryption during queries and inserts.

---

9. Asynchronous SQLAlchemy with Asyncio

• Using SQLAlchemy 2.0's asyncio support for modern async/await workflows.

• Integrating with asynchronous web frameworks like FastAPI.

---

10. Database Sharding

• Implementing sharding strategies with SQLAlchemy by distributing data across multiple databases.

• Managing connections and routing queries dynamically to the appropriate shard.

---

Would you like a detailed explanation or example for any of these?

---

3

Sure! Here are some more complex concepts of SQLAlchemy and Alembic, along with example code.

1. Handling Relationships and Cascade Options in SQLAlchemy

SQLAlchemy provides advanced features for handling relationships between models. You can define One-to-Many, Many-to-One, and Many-to-Many relationships, as well as apply cascading operations when performing operations like delete or update.

Code Example:

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Parent(Base):
    __tablename__ = 'parents'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    children = relationship('Child', back_populates='parent', cascade='all, delete-orphan')

class Child(Base):
    __tablename__ = 'children'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    parent_id = Column(Integer, ForeignKey('parents.id'))
    parent = relationship('Parent', back_populates='children')

# Set up the database and session
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Example operations
parent = Parent(name="John Doe")
child1 = Child(name="Child 1")
child2 = Child(name="Child 2")

parent.children = [child1, child2]

# Add parent and children to session and commit
session.add(parent)
session.commit()

# Now delete the parent, which should delete all associated children
session.delete(parent)
session.commit()

• Explanation:

  • cascade='all, delete-orphan' ensures that when a Parent object is deleted, all related Child objects are also deleted.

  • delete-orphan will remove any child that is removed from the relationship (orphaned).

2. Alembic Migrations: Handling Complex Schema Changes

When working with databases in production, schema changes need to be handled gracefully. Alembic helps in managing schema migrations without losing data.

Code Example:

Let’s create a migration for changing a column’s data type and adding constraints.

1. Initial Model:

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String, nullable=False)
    age = Column(Integer)

# Set up the database and session
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

1. Alembic Migration Script:

Use Alembic to create a migration script. First, run the following command to generate an initial migration script:

alembic revision --autogenerate -m "Initial migration"

This will create a migration script in migrations/versions/. Now, suppose we want to change the age column to String, which would require a manual update.

Edit the generated migration script like this:

def upgrade():
    op.alter_column('users', 'age', type_=String)

def downgrade():
    op.alter_column('users', 'age', type_=Integer)

• Explanation:

  • op.alter_column() is used to modify the age column’s data type from Integer to String.

  • In the downgrade() function, we specify how to revert the change.

1. Applying the Migration:

After modifying the migration script, run the following command to apply the changes:

alembic upgrade head

3. Advanced Querying: Subqueries and Complex Joins

SQLAlchemy allows the use of subqueries, complex joins, and filtering to build advanced queries.

Code Example:

from sqlalchemy import create_engine, Column, Integer, String, func, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Employee(Base):
    __tablename__ = 'employees'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    salary = Column(Integer)
    department_id = Column(Integer)

# Set up the database and session
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Adding sample data
session.add_all([
    Employee(name="Alice", salary=1000, department_id=1),
    Employee(name="Bob", salary=1500, department_id=1),
    Employee(name="Charlie", salary=2000, department_id=2),
])
session.commit()

# Subquery to find the highest salary in each department
subquery = (
    select([Employee.department_id, func.max(Employee.salary)])
    .group_by(Employee.department_id)
    .alias('max_salary')
)

# Main query to join the subquery and fetch employees with highest salaries
query = (
    session.query(Employee)
    .join(subquery, Employee.department_id == subquery.c.department_id)
    .filter(Employee.salary == subquery.c.max_salary)
)

result = query.all()
for employee in result:
    print(f"{employee.name} - {employee.salary}")

• Explanation:

  • We create a subquery to find the highest salary in each department using func.max().

  • Then, we use the main query to join the Employee table with the subquery and filter employees whose salary is equal to the maximum salary in their department.

4. Custom Types and Column Default Values

SQLAlchemy allows you to create custom column types and set complex default values for columns.

Code Example:

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.types import TypeDecorator

Base = declarative_base()

# Custom Type to store a list of strings
class StringList(TypeDecorator):
    impl = String

    def process_bind_param(self, value, dialect):
        if value is not None:
            return ",".join(value)
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            return value.split(",")
        return value

# Model using the custom type
class Product(Base):
    __tablename__ = 'products'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    tags = Column(StringList)

# Set up the database and session
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# Adding a product with a list of tags
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()

product = Product(name="Laptop", tags=["Electronics", "Gadgets", "Sale"])
session.add(product)
session.commit()

# Query the product and check the tags
retrieved_product = session.query(Product).first()
print(retrieved_product.tags)  # Output: ['Electronics', 'Gadgets', 'Sale']

• Explanation:

  • We created a custom column type StringList that stores a list of strings as a comma-separated string in the database.

  • process_bind_param converts the list to a string before storing it, and process_result_value splits the string back into a list when fetching it from the database.

5. Concurrency Handling and Locking

SQLAlchemy supports different isolation levels and locking mechanisms to handle concurrent access to database records.

Code Example:

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

Base = declarative_base()

class Item(Base):
    __tablename__ = 'items'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    quantity = Column(Integer)

# Set up the database and session
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Add sample data
session.add_all([Item(name="item1", quantity=10), Item(name="item2", quantity=5)])
session.commit()

# Example of locking a row to prevent race conditions
item_to_lock = session.query(Item).filter_by(name="item1").with_for_update().first()

# Simulate updating the item
item_to_lock.quantity -= 1
session.commit()

• Explanation:

  • with_for_update() locks the row during the transaction, which prevents other transactions from modifying the same record until the current transaction is committed.

Conclusion

These are some of the advanced features and concepts in SQLAlchemy and Alembic, including handling relationships, migrations, advanced querying, custom types, and concurrency management.

---

4

Here are 10 advanced SQLAlchemy and Alembic topics with examples:

---

1. Using SQLAlchemy hybrid_property for Custom Queries

Hybrid properties allow you to define attributes that can be used both in Python and in SQLAlchemy queries.

from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import Column, Integer

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

    @hybrid_property
    def full_name(self):
        return f"{self.first_name} {self.last_name}"

    @full_name.expression
    def full_name(cls):
        return cls.first_name + " " + cls.last_name

# Query
session.query(User).filter(User.full_name == "John Doe").all()

---

2. Composite Keys

Using composite primary keys for tables.

from sqlalchemy import Table, MetaData, Column, Integer, String, PrimaryKeyConstraint

metadata = MetaData()

orders = Table(
    "orders",
    metadata,
    Column("order_id", Integer),
    Column("product_id", Integer),
    Column("quantity", Integer),
    PrimaryKeyConstraint("order_id", "product_id")
)

---

3. SQLAlchemy Query with Subqueries

Perform advanced queries using subqueries.

from sqlalchemy import func, select

subquery = session.query(
    User.id, func.count(Post.id).label("post_count")
).join(Post).group_by(User.id).subquery()

result = session.query(subquery.c.id, subquery.c.post_count).all()

---

4. JSON Columns with SQLAlchemy

Store and query JSON data.

from sqlalchemy.dialects.postgresql import JSON

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True)
    data = Column(JSON)

# Query JSON keys
session.query(Product).filter(Product.data["category"].astext == "electronics").all()

---

5. Polymorphic Inheritance

Model table inheritance with polymorphism.

from sqlalchemy.orm import polymorphic_union

class Employee(Base):
    __tablename__ = "employees"
    id = Column(Integer, primary_key=True)
    type = Column(String)

    __mapper_args__ = {
        "polymorphic_identity": "employee",
        "polymorphic_on": type
    }

class Engineer(Employee):
    __tablename__ = "engineers"
    id = Column(Integer, ForeignKey("employees.id"), primary_key=True)
    field = Column(String)

    __mapper_args__ = {"polymorphic_identity": "engineer"}

# Querying polymorphic objects
session.query(Employee).all()

---

6. Alembic Environment Configuration with Multiple Databases

Set up Alembic for multiple database configurations.

# alembic.ini
[alembic]
script_location = alembic
sqlalchemy.url = sqlite:///main.db

# Add multiple URLs in env.py
from alembic import context

def run_migrations():
    context.configure(url="sqlite:///main.db")
    context.run_migrations()

    context.configure(url="sqlite:///secondary.db")
    context.run_migrations()

---

7. Conditional Indexes

Create partial indexes using SQLAlchemy.

from sqlalchemy.schema import Index

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

Index("idx_active_email", User.email, postgresql_where=User.is_active.is_(True))

---

8. Trigger and Event Handling

Handle database triggers using SQLAlchemy events.

from sqlalchemy.event import listen

def before_insert(mapper, connection, target):
    print(f"Before inserting: {target}")

listen(User, "before_insert", before_insert)

---

9. Alembic Custom Operations

Run custom SQL commands in Alembic migrations.

def upgrade():
    op.execute("CREATE TRIGGER update_timestamp BEFORE UPDATE ON users FOR EACH ROW EXECUTE FUNCTION update_timestamp();")

---

10. Bulk Insert and Update with ORM

Perform bulk operations for performance.

# Bulk Insert
session.bulk_insert_mappings(User, [
    {"id": 1, "name": "John"},
    {"id": 2, "name": "Jane"}
])

# Bulk Update
session.bulk_update_mappings(User, [
    {"id": 1, "name": "Johnny"},
    {"id": 2, "name": "Janet"}
])

---

Let me know if you’d like more examples!

---

5

---

6

---

7

---

8

---

9

---

10

---

11

---

12

---

13

---

14

---

15

---

16

---

17

---

18

---

19

---

20

---

21

---

22

---

23

---

24

---

25

---