DEV Community

Marwan Chawa
Marwan Chawa

Posted on

SQLModel & Alembic Tutorial

#sqlmodel #alembic #python #tutorial

This tutorial will guide you through the main concepts of SQLModel for defining and interacting with SQL databases, and Alembic for managing database schema migrations.

Part 1: Introduction to SQLModel

What is SQLModel?

SQLModel is a Python library that simplifies interaction with SQL databases. It cleverly combines the strengths of:

  • Pydantic: For data validation, settings management, and Python type hints. This means your data objects are well-defined and validated automatically.
  • SQLAlchemy: A powerful SQL toolkit and Object-Relational Mapper (ORM). This allows you to interact with your database using Python objects instead of writing raw SQL queries for many common tasks.

Key Benefits:

  • Intuitive: Write your data models once, and get both the ORM representation and the Pydantic model for validation and serialization.
  • Type Safety: Leverages Python type hints for better editor support (autocompletion, type checking) and fewer runtime errors.
  • Automatic Data Validation: Ensures that data conforms to your defined schema before it goes into the database.
  • SQLAlchemy Power: Access the full capabilities of SQLAlchemy when needed for complex queries or database operations.

Installation:

First, you'll need to install SQLModel. It's good practice to do this in a virtual environment.

pip install sqlmodel
Enter fullscreen mode Exit fullscreen mode

SQLModel doesn't come with a specific database driver by default. For this tutorial, we'll use SQLite, which is built into Python, so you won't need a separate database server.

Part 2: Defining Your First SQLModel

Let's define a simple model. Imagine we're creating an application to manage heroes. Our first model will be Hero. A model in SQLModel is just a Python class that inherits from SQLModel. If this model represents a database table, we add table=True to its internal Config class.

Here's the initial Hero model: 

import uuid
from typing import Optional

from sqlmodel import Field, SQLModel
from sqlalchemy import Column 
from sqlalchemy.types import JSON 
from pydantic import ConfigDict 

class Hero(SQLModel, table=True):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True, index=True, nullable=False)
    name: str = Field(index=True) 
    secret_name: str 
    age: Optional[int] = Field(default=None, index=True) 
    powers: Optional[list[str]] = Field(default=None, sa_column=Column(JSON))
    attributes_data: Optional[dict[str, any]] = Field(default=None, sa_column=Column(JSON))
Enter fullscreen mode Exit fullscreen mode

Explanation of the Hero model:

  • model_config = ConfigDict(arbitrary_types_allowed=True): This tells Pydantic it's okay to use any in dict[str, any], resolving potential schema generation errors.
  • id: uuid.UUID = ...: A UUID primary key, automatically generated by uuid.uuid4.
  • name: str = Field(index=True): Indexed for faster queries.
  • powers: Optional[list[str]] = Field(default=None, sa_column=Column(JSON)): Stores a list of strings as JSON. We use JSON here because many relational databases, including SQLite (which we're using in this tutorial), don't have a native column type for storing Python lists directly. Storing the list as a JSON string allows us to represent this structured data within a single database column. While some other database systems might offer native array types, using JSON provides a widely compatible approach.
  • attributes_data: Optional[dict[str, any]] = Field(default=None, sa_column=Column(JSON)): Stores a dictionary as JSON. Similar to lists, Python dictionaries are not a native type in SQLite. Serializing the dictionary to a JSON string allows us to store this key-value data. This is especially useful for flexible, semi-structured data where the keys or structure might vary. Other databases might have native JSONB or map types that offer more advanced querying capabilities, but JSON is a good general-purpose solution.

Commonly Used Practices in Defining Columns

When defining columns for your SQLModel tables, several common practices and Field parameters are frequently used to ensure your data is well-structured, efficient, and behaves as expected.

  1. Type Hinting is Key (including datetime):

    • Always use Python type hints for your model fields (e.g., str, int, bool, uuid.UUID, datetime.datetime). SQLModel uses these hints to determine the appropriate database column type.
    • For optional fields, use Optional[YourType] and provide a default value, often None.
    • For datetime fields, it's common to set a default value to the current time when a record is created. 
    import datetime # Make sure datetime is imported
import uuid     # Make sure uuid is imported
from typing import Optional

name: str  # Required string
age: Optional[int] = Field(default=None) # Optional integer

# Datetime field, automatically set to current UTC time on creation
created_at: datetime.datetime = Field(default_factory=datetime.datetime.utcnow, nullable=False)

# Optional datetime field
updated_at: Optional[datetime.datetime] = Field(default=None)

  2. Primary Keys (primary_key=True) (including Auto-increment):

    • Every table should have a primary key. This uniquely identifies each row.
    • Common choices are an auto-incrementing integer or a UUID.
    • nullable=False is implicit for primary keys. index=True is also typically automatically applied by the database for primary keys. 
    import uuid # Make sure uuid is imported
from typing import Optional

# Auto-incrementing integer primary key
# SQLModel/SQLAlchemy handle the "AUTOINCREMENT" or "SERIAL" behavior for supported databases
# when primary_key=True and the type is Optional[int] with default=None.
item_id: Optional[int] = Field(default=None, primary_key=True)

# UUID primary key (recommended for distributed systems or when you want to generate IDs in application code)
user_id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)

*   Note: While `index=True` is often automatically applied to primary keys by the database, explicitly stating `index=True` in `Field` for primary keys doesn't hurt and can make the intent clearer, though it's often redundant.
Enter fullscreen mode Exit fullscreen mode

  1. Default Values (default and default_factory):

    • default: Use for static default values (e.g., default=0, default=True).
    • default_factory: Use for dynamic default values that need to be generated each time a new record is created (e.g., default_factory=uuid.uuid4, default_factory=datetime.datetime.utcnow). 
    is_active: bool = Field(default=True)
last_login: Optional[datetime.datetime] = Field(default=None)
unique_code: str = Field(default_factory=lambda: secrets.token_hex(8)) # Using a lambda for a simple factory

  2. Indexing (index=True):

    • Add index=True to columns that are frequently used in WHERE clauses of your queries, or for sorting (ORDER BY).
    • This significantly speeds up read operations on those columns.
    • Be mindful that indexes add a small overhead to write operations (INSERT, UPDATE, DELETE). 
    email: str = Field(index=True, unique=True) # Often indexed and unique
last_name: str = Field(index=True)

  3. Nullability (nullable=False or Optional):

    • By default, fields are required (not nullable).
    • To make a field nullable (allow NULL values in the database), type hint it with Optional[YourType] and ensure it has a default (often None) or default_factory.
    • If a field must have a value, you can explicitly set nullable=False in Field, though this is often inferred if the type is not Optional and no default is provided. For primary keys, nullable=False is implicit. 
    description: Optional[str] = Field(default=None) # Allows NULL
# For a required field, just the type hint is enough, or be explicit:
# product_code: str = Field(nullable=False) # This is often the default if not Optional

  4. Uniqueness (unique=True):

    • If a column's value must be unique across all rows in the table (e.g., email address, username), set unique=True in Field.
    • This creates a unique constraint in the database. 
    username: str = Field(unique=True, index=True) # Usernames are typically unique and indexed

  5. String Lengths and Specific Types (sa_column or sa_type):

    • For str fields, the default database type might be TEXT or VARCHAR without a specific length. If you need to enforce a maximum length or use a specific database type, use sa_column or sa_type. 
    from sqlalchemy import String, Text, Column
from sqlalchemy.types import JSON # Already imported in the example above

# Using sa_type for specific string length
short_code: str = Field(sa_type=String(10))

# Using sa_column for more complex definitions or when SQLModel's default isn't quite right
# (e.g., JSON columns as shown in the Hero model, or specific constraints)
# powers: Optional[list[str]] = Field(default=None, sa_column=Column(JSON))
# notes: Optional[str] = Field(default=None, sa_column=Column(Text))

  6. Relationships (Foreign Keys):

    • To define relationships between tables, you'll use ForeignKey. This is typically done by passing foreign_key to Field.
    • The value for foreign_key should be a string like "other_table_name.other_table_primary_key_column_name". 
    # Example (assuming a Team model exists)
# team_id: Optional[int] = Field(default=None, foreign_key="team.id")

*   SQLModel also helps define the relationship attributes for easy access to related objects (e.g., `team: Optional["Team"] = Relationship(back_populates="heroes")`). This will be covered in more detail when discussing relationships.
Enter fullscreen mode Exit fullscreen mode

By applying these practices, you can create robust and efficient database schemas with SQLModel.

Defining Relationships Between Models

SQLModel, leveraging SQLAlchemy's capabilities, allows you to define relationships between your database tables. This is crucial for modeling real-world connections, like a hero belonging to a team, or a product having multiple reviews.

The two main components for defining relationships are:

  1. Foreign Keys: A field in one table that refers to the primary key of another table. This establishes the link at the database level.
  2. Relationship Attributes: Python attributes on your SQLModel classes that allow you to navigate these connections as if they were regular object attributes (e.g., hero.team or team.heroes). These are defined using Relationship from sqlmodel.

Let's explore common relationship types:

1. One-to-Many / Many-to-One Relationships

This is the most common type. For example, one Team can have many Hero objects, and each Hero belongs to at most one Team.

a. Define the "One" side (e.g., Team model):
The "one" side will have a collection of the "many" side.

# models.py (example)
import uuid
from typing import Optional, list
from sqlmodel import Field, SQLModel, Relationship # Add Relationship

class Team(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    name: str = Field(unique=True, index=True)
    headquarters: str

    # This creates the 'team.heroes' attribute
    # The 'back_populates' tells SQLAlchemy which attribute on the Hero model
    # refers back to this relationship.
    heroes: list["Hero"] = Relationship(back_populates="team")

# Forward declaration for Hero if it's in the same file or imported later
# class Hero(SQLModel, table=True): ...
Enter fullscreen mode Exit fullscreen mode

b. Define the "Many" side (e.g., Hero model):
The "many" side will have a foreign key field pointing to the "one" side, and a relationship attribute to access the "one" side.

# models.py (continued)

class Hero(SQLModel, table=True):
    # ... (id, name, secret_name, age, etc. from previous examples)
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

    # Foreign key to the Team table's id
    team_id: Optional[uuid.UUID] = Field(default=None, foreign_key="team.id")

    # This creates the 'hero.team' attribute
    # 'back_populates' links it to the 'heroes' attribute on the Team model.
    team: Optional["Team"] = Relationship(back_populates="heroes")
Enter fullscreen mode Exit fullscreen mode

Explanation:

  • team_id: Optional[uuid.UUID] = Field(default=None, foreign_key="team.id"):
    • This field in the Hero model is the foreign key.
    • foreign_key="team.id" tells SQLAlchemy that this column should reference the id column of the team table.
    • It's Optional because a hero might not belong to a team. If a hero must belong to a team, you would make it non-optional (e.g., team_id: uuid.UUID = Field(foreign_key="team.id")).
  • heroes: list["Hero"] = Relationship(back_populates="team") (in Team model):
    • This defines the team.heroes attribute. When you have a Team object, you can access my_team.heroes to get a list of Hero objects associated with that team.
    • back_populates="team" specifies that the team attribute on the Hero model is the other side of this relationship.
  • team: Optional["Team"] = Relationship(back_populates="heroes") (in Hero model):
    • This defines the hero.team attribute. When you have a Hero object, you can access my_hero.team to get the Team object this hero belongs to.
    • back_populates="heroes" specifies that the heroes attribute on the Team model is the other side of this relationship.
  • Type Hinting with Strings ("Hero", "Team"):
    • The type hints for relationship attributes (list["Hero"] and Optional["Team"]) use strings. This is called a "forward reference." It's necessary when the related class (Hero or Team) hasn't been fully defined yet in the Python file, or to avoid circular import errors if models are in different files. SQLModel (via Pydantic) resolves these string hints later.

How it works in practice:

# (Assuming engine and session are set up)
# team_avengers = Team(name="Avengers", headquarters="Avengers Tower")
# hero_ironman = Hero(name="Iron Man", secret_name="Tony Stark", team=team_avengers) # Link directly
# hero_hulk = Hero(name="Hulk", secret_name="Bruce Banner", team_id=team_avengers.id) # Or link via FK

# session.add(team_avengers)
# session.add(hero_ironman)
# session.add(hero_hulk)
# session.commit()

# retrieved_team = session.get(Team, team_avengers.id)
# for hero in retrieved_team.heroes:
#     print(f"{hero.name} is in {retrieved_team.name}")

# retrieved_hero = session.get(Hero, hero_ironman.id)
# print(f"{retrieved_hero.name}'s team is {retrieved_hero.team.name}")
Enter fullscreen mode Exit fullscreen mode

2. Many-to-Many Relationships

Sometimes, entities can be related to many other entities, and vice-versa. For example, a Hero can have many Powers, and a Power can be possessed by many Heroes. This requires an "association table" (also known as a link table or join table).

Let's imagine we have Hero and Power models.

a. Define the main models (Hero, Power):

# models.py (continued)
class Power(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    name: str = Field(unique=True, index=True)
    description: Optional[str] = None

    # Relationship to the association table (HeroPowerLink)
    # and then through it to Hero
    heroes: list["Hero"] = Relationship(back_populates="powers", link_model=HeroPowerLink)

# We need an association model for the many-to-many relationship
class HeroPowerLink(SQLModel, table=True):
    hero_id: uuid.UUID = Field(default=None, primary_key=True, foreign_key="hero.id")
    power_id: uuid.UUID = Field(default=None, primary_key=True, foreign_key="power.id")

# Update Hero model to include relationship to Power
class Hero(SQLModel, table=True): # Assuming Hero is already defined above
    # ... existing fields ...
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) # ensure it's here for clarity
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)
    team_id: Optional[uuid.UUID] = Field(default=None, foreign_key="team.id")
    team: Optional["Team"] = Relationship(back_populates="heroes")

    # Relationship to the association table (HeroPowerLink)
    # and then through it to Power
    powers: list["Power"] = Relationship(back_populates="heroes", link_model=HeroPowerLink)
Enter fullscreen mode Exit fullscreen mode

Explanation of Many-to-Many:

  • HeroPowerLink (Association Table):
    • This model represents the table that links Hero and Power.
    • It has two fields, hero_id and power_id, which are foreign keys to hero.id and power.id respectively.
    • Both hero_id and power_id together form a composite primary key for this table to ensure each hero-power link is unique.
  • link_model=HeroPowerLink:
    • In the Relationship definitions on both Hero (for powers) and Power (for heroes), the link_model parameter tells SQLModel to use HeroPowerLink as the intermediary table for this many-to-many relationship.
  • back_populates:
    • Works similarly to one-to-many, ensuring the relationship is bi-directional. hero.powers links to power.heroes.

How it works in practice:

# (Assuming engine and session are set up)
# power_flight = Power(name="Flight", description="Ability to fly")
# power_strength = Power(name="Super Strength", description="Enhanced physical strength")

# hero_superman = Hero(name="Superman", secret_name="Clark Kent")
# hero_wonderwoman = Hero(name="Wonder Woman", secret_name="Diana Prince")

# # Associate powers with heroes
# hero_superman.powers.append(power_flight)
# hero_superman.powers.append(power_strength)
# hero_wonderwoman.powers.append(power_strength)
# # SQLModel handles creating the HeroPowerLink records automatically when you commit

# session.add_all([power_flight, power_strength, hero_superman, hero_wonderwoman])
# session.commit()

# retrieved_superman = session.get(Hero, hero_superman.id)
# print(f"{retrieved_superman.name} has powers:")
# for power in retrieved_superman.powers:
#     print(f"- {power.name}")

# retrieved_strength = session.get(Power, power_strength.id)
# print(f"Heroes with {retrieved_strength.name}:")
# for hero in retrieved_strength.heroes:
#     print(f"- {hero.name}")
Enter fullscreen mode Exit fullscreen mode

Important Considerations for Relationships:

  • Lazy Loading vs. Eager Loading: By default, related objects (hero.team, team.heroes) are "lazy-loaded." This means the database query to fetch them is only executed when you first access the attribute. For performance-critical sections, especially when dealing with lists of objects, you might want to "eagerly load" these relationships (fetch them in the initial query). This can be done using SQLAlchemy's loader options (e.g., selectinload) with session.exec(). This is a more advanced topic.

    # Example of eager loading (more advanced)
# from sqlalchemy.orm import selectinload
# statement = select(Team).options(selectinload(Team.heroes))
# teams_with_heroes = session.exec(statement).all()

  • Cascade Operations: You can configure what happens to related objects when an object is deleted (e.g., delete all heroes if their team is deleted). This is handled by SQLAlchemy's cascade options, often configured in Relationship.

  • Circular Imports: If your models are in separate files, be careful with import statements. Using forward references (string type hints like "Team") helps. You might also need to adjust where you import models or use if TYPE_CHECKING: blocks.

Defining relationships correctly is key to building a well-structured and maintainable application with SQLModel.

Understanding index=True:

When you specify index=True (e.g., name: str = Field(index=True)), the database creates an index on that column.

How it Works:

  • An index is like a book's index, a lookup table for fast data retrieval. It helps avoid full table scans.

Benefits:

  • Faster queries (filtering with WHERE, sorting with ORDER BY, joins).
  • The database optimizer uses indexes automatically.
  • Primary keys are usually indexed by default.

Considerations:

  • Slight overhead on writes (INSERT, UPDATE, DELETE) as indexes also need updating.
  • Indexes consume disk space.
  • Most effective on columns with high selectivity (many unique values).

Important Note on JSON fields:

Querying within JSON structures might require database-specific functions. For storing/retrieving structured data, JSON fields are very useful.

Advanced SQLAlchemy Column Customization with Field parameters:

SQLModel's Field function is the bridge for configuring the underlying SQLAlchemy Column for each model attribute. To achieve fine-grained control, you can pass several sa_* prefixed parameters to Field. These parameters are passed directly to SQLAlchemy, allowing you to customize aspects of the column definition that go beyond standard SQLModel behavior. Here are the key ones:

  • sa_column:

    • Purpose: Allows you to provide a fully customized SQLAlchemy Column object.
    • Example: 
    sa_column=Column(JSON, nullable=True)
    • This explicitly defines the column as a JSON type and allows NULL values.
    • Use Case: Use this when you need to customize the column's behavior or type beyond what SQLModel's default mapping provides. For instance, defining a JSON column or adding specific constraints not covered by other Field parameters.

  • sa_type:

    • Purpose: Lets you specify only the SQLAlchemy type for the column, if SQLModel's automatic type inference isn't sufficient or if you need a very specific database type.
    • Example: 
    sa_type=String(30)
    • This defines the column as a string with a maximum length of 30 characters (e.g., VARCHAR(30) in SQL).
    • Use Case: Use this when you want to control the SQL type (e.g., String, Integer, DateTime, or database-specific types) but don't need to define a full Column object with all its arguments.

  • sa_column_args:

    • Purpose: Passes a tuple of positional arguments directly to the SQLAlchemy Column constructor. This is often used for things like ForeignKey constraints.
    • Example: 
    sa_column_args=(ForeignKey("other_table.id"),)
    • This sets up a foreign key relationship from this column to the id column of other_table.
    • Use Case: Use this when you need to add specific constraints or relationships to the column that are passed as positional arguments to SQLAlchemy's Column (primarily for foreign keys, but potentially others depending on Column extensions).

  • sa_column_kwargs:

    • Purpose: Passes a dictionary of keyword arguments directly to the SQLAlchemy Column constructor. This allows access to any Column parameter not covered by other Field options.
    • Examples:
    • sa_column_kwargs={"server_default": text("gen_random_uuid()")}
      • Sets a server-side default value. The database will execute the provided SQL expression (here, calling a function like gen_random_uuid()) if no value is provided for this field upon insertion. (Note: text is typically imported: from sqlalchemy import text).
    • sa_column_kwargs={"onupdate": func.now()}
      • Automatically updates the field to the current timestamp (using the database's NOW() function or equivalent) whenever the record is updated in the database. (Note: func is typically imported: from sqlalchemy import func).
    • sa_column_kwargs={"comment": "User's primary email address"}
      • Adds a comment to the column in the database schema, which can be useful for documentation and is supported by some database systems.
    • Use Case: Use this for advanced behaviors like server-side defaults, on-update triggers, column comments, specifying server-side functions, or any other specific SQLAlchemy Column keyword arguments.

Why Use These Parameters?

These parameters provide fine-grained control over how your database columns are created and behave. While SQLModel provides sensible defaults, there are scenarios where you need to:

  • Integrate with existing database schemas.
  • Use advanced database features like triggers, constraints, or custom types.
  • Optimize performance or enforce specific rules at the database level.

By combining these parameters, you can fully leverage SQLAlchemy's power while still benefiting from SQLModel's simplicity.

Code at the end of Part 2 (main.py): 

import uuid
from typing import Optional

from sqlmodel import Field, SQLModel
from sqlalchemy import Column
from sqlalchemy.types import JSON
from pydantic import ConfigDict

class Hero(SQLModel, table=True):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True, index=True, nullable=False)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)
    powers: Optional[list[str]] = Field(default=None, sa_column=Column(JSON))
    attributes_data: Optional[dict[str, any]] = Field(default=None, sa_column=Column(JSON))

if __name__ == "__main__":
    # We will add database interaction code here in later parts
    print("Hero model defined.")
Enter fullscreen mode Exit fullscreen mode

Part 3: Creating the Database and Tables

With the model defined, we set up the database.

1. Database URL:

For SQLite, this is the path to the database file.

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
Enter fullscreen mode Exit fullscreen mode

2. Create Engine:

The engine is the entry point to the database. echo=True logs SQL statements.

from sqlmodel import create_engine  # Add create_engine to imports if not already there
engine = create_engine(sqlite_url, echo=True)
Enter fullscreen mode Exit fullscreen mode

3. Create Tables Function:

SQLModel.metadata.create_all(engine) creates tables defined by SQLModel models if they don't exist.

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)
Enter fullscreen mode Exit fullscreen mode

Caveat: create_all(engine) will not alter existing tables. If you change a model (e.g., add a field) after the table exists, the schema won't update. This is where Alembic (for migrations) becomes crucial. To see schema changes with create_all during initial development, you might delete database.db first.

Code at the end of Part 3 (main.py): 

import uuid
from typing import Optional

from sqlmodel import Field, SQLModel, create_engine  # Added create_engine
from sqlalchemy import Column
from sqlalchemy.types import JSON
from pydantic import ConfigDict

# --- Model Definition (from Part 2) ---
class Hero(SQLModel, table=True):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True, index=True, nullable=False)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)
    powers: Optional[list[str]] = Field(default=None, sa_column=Column(JSON))
    attributes_data: Optional[dict[str, any]] = Field(default=None, sa_column=Column(JSON))

# --- Database Setup (Part 3) ---
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

# Main execution block
if __name__ == "__main__":
    create_db_and_tables()
    print("Database and tables created (if they didn't exist)!")
    # CRUD operations will be added in Part 4
Enter fullscreen mode Exit fullscreen mode

Part 4: Interacting with Data (CRUD Operations)

CRUD stands for Create, Read, Update, and Delete. SQLModel uses a Session to manage database interactions.

1. The Session:

A Session (imported from sqlmodel) is like a temporary workspace for your database objects. You load objects into it, make changes, and then commit() these changes to the database or rollback() to discard them. It's best practice to use a session within a with statement to ensure it's properly closed:

from sqlmodel import Session  # Make sure Session is imported
# ...
# with Session(engine) as session:
#     # Your database operations go here
#     pass
Enter fullscreen mode Exit fullscreen mode

All database operations (creating, reading, updating, deleting) happen within the context of a session.

2. Creating Data (Create):

To add new records to your database:

  • Instantiate your SQLModel class: Create an instance of your model with the data for the new record (e.g., hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")). The id field (if it's a primary key with a default or default factory) will usually be handled automatically.
  • Add the instance to the session: Use session.add(hero_1) to place the new object into the session's tracking.
  • Commit the session: Call session.commit(). This flushes all pending changes (inserts, updates, deletes) to the database and makes them permanent.
  • Refresh the object (optional but good practice): After committing, if the database generated any values for your object (like an auto-incrementing ID, or a default value set by the database), your Python object won't automatically have these. session.refresh(hero_1) reloads the object's state from the database, ensuring your Python instance is up-to-date. For UUIDs generated by default_factory in Python, the ID is known before the commit, but refresh is still good for other DB-generated values.

3. Reading Data (Read):

To fetch data from the database:

  • Use select(): Import select from sqlmodel. To select all heroes, you'd write statement = select(Hero).
  • Filter with .where(): To add conditions, use the .where() method.
    • Basic Equality: select(Hero).where(Hero.name == "Spider-Boy")
    • Comparisons: select(Hero).where(Hero.age > 30), select(Hero).where(Hero.age <= 18)
    • Multiple Conditions (AND):
      • Chaining .where(): select(Hero).where(Hero.age > 18).where(Hero.name == "Spider-Boy")
      • Using and_ (requires from sqlalchemy import and_): select(Hero).where(and_(Hero.age > 18, Hero.name == "Spider-Boy"))
    • Multiple Conditions (OR) (requires from sqlalchemy import or_):
      • select(Hero).where(or_(Hero.age < 18, Hero.age > 60))
    • Checking for None (NULL values):
      • select(Hero).where(Hero.age == None) (for columns that can be NULL)
      • select(Hero).where(Hero.age != None) (for columns that are NOT NULL)
    • IN clause (matching one of several values):
      • select(Hero).where(Hero.name.in_(["Spider-Boy", "Deadpond", "Rusty-Man"]))
    • NOT IN clause:
      • select(Hero).where(Hero.name.not_in_(["Captain North", "Doctor South"]))
    • String matching (like, ilike):
      • select(Hero).where(Hero.name.like("Spider%")) (case-sensitive like)
      • select(Hero).where(Hero.name.ilike("spider%")) (case-insensitive like - behavior might depend on DB)
    • Combining with other model fields: You can compare fields within the same model or across joined models (more on joins later).
  • Execute the query: Use results = session.exec(statement). This executes the SQL query against the database.
  • Fetch results:
    • heroes_list = results.all(): Returns a list of all matching model instances.
    • one_hero = results.first(): Returns the first matching model instance, or None if no records match.
    • one_hero = results.one(): Returns exactly one model instance. If zero or more than one record matches, it raises an error. This is useful when you expect a unique result.
    • one_hero = session.get(Hero, hero_id): If you know the primary key of an object, session.get() is an efficient way to retrieve it directly.

4. Updating Data (Update):

To modify existing records:

  • Fetch the object: First, read the object you want to update from the database using one of the methods described above (e.g., hero_to_update = session.exec(select(Hero).where(Hero.name == "Spider-Boy")).first()).
  • Modify attributes: Change the attributes of the fetched Python object directly (e.g., hero_to_update.age = 17).
  • Handle mutable types: If you modify a mutable type in-place (like appending to a list in a JSON field, e.g., hero_to_update.powers.append("new_power")), SQLAlchemy's change tracking usually picks this up. However, to be absolutely certain, especially if only the mutable field changed, you can explicitly mark the attribute as modified using flag_modified from sqlalchemy.orm.attributes: 
from sqlalchemy.orm.attributes import flag_modified
# ...
# flag_modified(hero_to_update, "powers")
Enter fullscreen mode Exit fullscreen mode
  • Add to session (if needed): If the object was fetched within the current session, it's already tracked. Modifying it makes it "dirty." session.add(hero_to_update) can be used; if the object is already persistent and tracked, this re-associates it if needed and ensures it's considered for the commit.
  • Commit the session: session.commit() saves the changes to the database by issuing an UPDATE statement.
  • Refresh (optional): session.refresh(hero_to_update) can be used to get the latest state from the database, though often not strictly necessary after an update unless there are database-side triggers or defaults affecting other fields.

5. Deleting Data (Delete):

To remove records from the database:

  • Fetch the object: First, read the object you want to delete (e.g., hero_to_delete = session.exec(select(Hero).where(Hero.name == "Rusty-Man")).first()).
  • Delete from session: Use session.delete(hero_to_delete). This marks the object for deletion.
  • Commit the session: session.commit() issues the DELETE statement to the database. After the commit, the Python object hero_to_delete is in a "deleted" state and typically shouldn't be used further.

Code at the end of Part 4 (main.py): 

import uuid
from typing import Optional

from sqlmodel import Field, SQLModel, create_engine, Session, select  # Added Session, select
from sqlalchemy import Column
from sqlalchemy.types import JSON
from sqlalchemy.orm.attributes import flag_modified  # Import for updating JSON lists
from pydantic import ConfigDict

# --- Model Definition (from Part 2) ---
class Hero(SQLModel, table=True):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True, index=True, nullable=False)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)
    powers: Optional[list[str]] = Field(default=None, sa_column=Column(JSON))
    attributes_data: Optional[dict[str, any]] = Field(default=None, sa_column=Column(JSON))
    # allegiance field is not yet added to the model at this stage of the tutorial

# --- Database Setup (Part 3) ---
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

# --- CRUD Functions (Part 4) ---
def create_heroes_example():
    print("\n--- Creating heroes ---")
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", powers=["healing", "teleportation"], attributes_data={"suit_color": "red", "humor_level": 10})
    hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador", age=16, powers=["web-slinging", "wall-crawling"], attributes_data={"mentor": "Deadpond", "school": "Midtown High"})
    hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48, powers=["genius intellect", "powered armor"], attributes_data={"company": "Sharp Industries", "armor_mark": "XLII"})

    with Session(engine) as session:
        session.add(hero_1)
        session.add(hero_2)
        session.add(hero_3)
        session.commit()

        session.refresh(hero_1)
        session.refresh(hero_2)
        session.refresh(hero_3)
        print(f"Created Hero: ID={hero_1.id}, Name={hero_1.name}")
        print(f"Created Hero: ID={hero_2.id}, Name={hero_2.name}")
        print(f"Created Hero: ID={hero_3.id}, Name={hero_3.name}")

def read_heroes_example():
    print("\n--- Reading heroes ---")
    with Session(engine) as session:
        statement = select(Hero)
        heroes = session.exec(statement).all()
        print("All heroes in database:")
        for hero in heroes:
            print(f"  ID: {hero.id}, Name: {hero.name}, Age: {hero.age}, Powers: {hero.powers}, Attributes: {hero.attributes_data}")

        print("\nReading heroes older than 30:")
        statement_older = select(Hero).where(Hero.age > 30)
        heroes_older = session.exec(statement_older).all()
        for hero in heroes_older:
            print(f"  Name: {hero.name}, Age: {hero.age}")

        print("\nReading hero 'Deadpond':")
        statement_deadpond = select(Hero).where(Hero.name == "Deadpond")
        deadpond = session.exec(statement_deadpond).first()
        if deadpond:
            print(f"  Found: {deadpond.name}, Secret Name: {deadpond.secret_name}")
        else:
            print("  Deadpond not found.")

def update_heroes_example():
    print("\n--- Updating heroes ---")
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Spider-Boy")
        spider_boy = session.exec(statement).first()

        if spider_boy:
            print(f"Before update - Spider-Boy: Age={spider_boy.age}, Powers={spider_boy.powers}")
            spider_boy.age = 17 
            if spider_boy.powers: 
                 spider_boy.powers.append("super-strength (mild)")
            else:
                 spider_boy.powers = ["super-strength (mild)"]

            flag_modified(spider_boy, "powers")

            session.add(spider_boy) 
            session.commit() 
            session.refresh(spider_boy)
            print(f"After update - Spider-Boy: Age={spider_boy.age}, Powers={spider_boy.powers}")
        else:
            print("Spider-Boy not found for update.")

def delete_heroes_example():
    print("\n--- Deleting heroes ---")
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Rusty-Man")
        rusty_man = session.exec(statement).first()

        if rusty_man:
            print(f"Deleting hero: {rusty_man.name}")
            session.delete(rusty_man)
            session.commit()
            print(f"{rusty_man.name} deleted.")

            verify_statement = select(Hero).where(Hero.name == "Rusty-Man")
            deleted_hero = session.exec(verify_statement).first()
            if not deleted_hero:
                print(f"Verified: {rusty_man.name} is no longer in the database.")
            else:
                print(f"Verification FAILED: {rusty_man.name} still found.")
        else:
            print("Rusty-Man not found for deletion.")

# Main execution block
if __name__ == "__main__":
    create_db_and_tables()  # For initial setup if not using Alembic for the very first creation
    print("Database and tables created (if they didn't exist).")

    print("\nRunning CRUD examples...")
    create_heroes_example()
    read_heroes_example()
    update_heroes_example()
    delete_heroes_example()
    print("\n--- Reading heroes after all operations ---")
    read_heroes_example()
Enter fullscreen mode Exit fullscreen mode

Part 5: Database Migrations with Alembic

When your application evolves, your database schema (the structure of your tables, columns, etc.) will likely need to change. SQLModel.metadata.create_all(engine) does not handle schema modifications on existing databases. This is where Alembic comes in.

What is Alembic?

Alembic is a database migration tool for SQLAlchemy. It allows versioning schema changes, generating migration scripts, applying migrations, and working in teams.

1. Installation: 

pip install alembic
Enter fullscreen mode Exit fullscreen mode

2. Initializing Alembic Environment:

In your project's root directory, run:

alembic init alembic
Enter fullscreen mode Exit fullscreen mode

This creates an alembic directory and an alembic.ini file. Inside the alembic directory, you'll also find a script.py.mako file, which is the template for new migration scripts.

3. Configuring Alembic for SQLModel:

a. Configure alembic.ini:

Open alembic.ini. Set sqlalchemy.url:

sqlalchemy.url = sqlite:///./database.db
Enter fullscreen mode Exit fullscreen mode

(Adjust path if database.db is not in the project root relative to where you run Alembic commands).

b. Configure env.py:

Modify alembic/env.py:

  • Import your models' metadata and set target_metadata.

Near the top:

# alembic/env.py - near the top
# Assuming your models (like Hero) are in main.py:
import sys
import os
sys.path.insert(0, os.path.realpath(os.path.join(os.path.dirname(__file__), '..')))  # Add project root to sys.path
from main import Hero  # Or your_models_file.model_name
from sqlmodel import SQLModel
Enter fullscreen mode Exit fullscreen mode

Find target_metadata = None and change it to:

# alembic/env.py - inside run_migrations_offline() and run_migrations_online()
target_metadata = SQLModel.metadata
Enter fullscreen mode Exit fullscreen mode

The sys.path.insert line helps Alembic find your main.py (or models file) if it's in the parent directory of alembic/.

c. Configure script.py.mako (Optional - Proactive step for smoother autogeneration):

The alembic/script.py.mako file is a template used by Alembic to generate new migration scripts. Sometimes, Alembic's autogenerate feature might output type definitions that require specific imports (e.g., types from sqlmodel) to be present in the migration script for them to be resolved correctly.

You can proactively add common imports like import sqlmodel to the top of your script.py.mako file. This may help Alembic resolve certain SQLModel-specific types if it attempts to use them directly during autogeneration, potentially leading to cleaner initial migration scripts.

Open alembic/script.py.mako. At the top, where other imports are (like sqlalchemy and alembic), you can add an import:

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | render_down_revision}
Create Date: ${create_date}

"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# Example: Add sqlmodel import if autogenerate tends to use its types directly and they are resolvable
import sqlmodel 

# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}


def upgrade():
    ${upgrades if upgrades else "pass"}


def downgrade():
    ${downgrades if downgrades else "pass"}
Enter fullscreen mode Exit fullscreen mode

While this can be helpful, it's important to understand that the primary goal is for autogenerated scripts to use standard SQLAlchemy types (e.g., sa.String, sa.JSON). If Alembic autogenerates an internal SQLModel type that isn't meant for direct use (like sqlmodel.sql.sqltypes.AutoString), simply importing sqlmodel might not fix an AttributeError for that specific type. In such cases, manual correction to the standard SQLAlchemy type in the generated script is still the most direct solution. However, having import sqlmodel in the template can be beneficial if Alembic correctly identifies a SQLModel-provided type that is intended for use and just needs the import to be resolved.

d. Configure Naming Conventions and Batch Mode for SQLite (Recommended):

To reduce spurious differences detected by autogenerate (which can lead to empty migration scripts) and to better handle SQLite's limitations with ALTER TABLE, it's good practice to define naming conventions and enable batch mode in env.py.

In alembic/env.py, within the run_migrations_online() function, modify the context.configure call:

# Inside run_migrations_online() in env.py
# ...
# This is a common naming convention dictionary
naming_convention = {
    "ix": "ix_%(table_name)s_%(column_0_label)s",  # Added table_name for more uniqueness
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",  # Used constraint_name for check constraints
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s"
}

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    compare_type=True,  # Detects column type changes
    naming_convention=naming_convention,  # Add naming convention
    render_as_batch=True  # Enable batch mode, crucial for SQLite
    # ... other options
)
# ...
Enter fullscreen mode Exit fullscreen mode

Make similar changes to context.configure in run_migrations_offline() if you use offline mode. The render_as_batch=True is particularly important for SQLite. compare_type=True helps detect column type changes.

4. Creating an Initial Migration (Autogenerate):

Your first migration will capture the initial state of your schema based on the Hero model (without allegiance).

alembic revision -m "create_hero_table" --autogenerate
Enter fullscreen mode Exit fullscreen mode

Review the generated script in alembic/versions/ to ensure it correctly creates the hero table with id, name, secret_name, age, powers, and attributes_data. Then apply it:

alembic upgrade head
Enter fullscreen mode Exit fullscreen mode

5. Making a Schema Change and Generating a New Migration:

Now, we'll modify our Hero model to add the allegiance field.

First, update your Hero model definition in main.py (or your models file) to include the new allegiance field:

# In main.py (or your models file)
import uuid
from typing import Optional

from sqlmodel import Field, SQLModel
from sqlalchemy import Column
from sqlalchemy.types import JSON
from pydantic import ConfigDict

class Hero(SQLModel, table=True):  # Ensure this is the definition Alembic sees
    model_config = ConfigDict(arbitrary_types_allowed=True)

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True, index=True, nullable=False)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)
    powers: Optional[list[str]] = Field(default=None, sa_column=Column(JSON))
    attributes_data: Optional[dict[str, any]] = Field(default=None, sa_column=Column(JSON))
    allegiance: Optional[str] = Field(default=None, index=True)  # <<<--- ADD THIS LINE
Enter fullscreen mode Exit fullscreen mode

Save the file where your Hero model is defined.

a. Generate a new migration script:

Run the alembic revision command again, but with a new message describing the change:

alembic revision -m "add_allegiance_to_hero" --autogenerate
Enter fullscreen mode Exit fullscreen mode

Alembic will compare your updated SQLModel.metadata (which now includes the allegiance field in the Hero model) against the current state of the database (as recorded by the last applied migration). It should detect that a new column needs to be added. A new migration script will be created in alembic/versions/.

b. Review and Correct the New Migration Script:

Open the newly generated script (e.g., alembic/versions/<new_hash>_add_allegiance_to_hero.py).

You should see operations like op.add_column():

# Example content of the new migration script's upgrade() function:
def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('hero', sa.Column('allegiance', sa.String(), nullable=True))
    op.create_index(op.f('ix_hero_allegiance'), 'hero', ['allegiance'], unique=False)
    # ### end Alembic commands ###

# Example content of the new migration script's downgrade() function:
def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_hero_allegiance'), table_name='hero')
    op.drop_column('hero', 'allegiance')
    # ### end Alembic commands ###
Enter fullscreen mode Exit fullscreen mode
  • Verify Column Type: Ensure sa.String() is appropriate.
  • Verify Nullable: nullable=True is correct since allegiance is Optional[str].
  • Verify Index: op.create_index(...) is correct because we added index=True to the allegiance field.
  • Remove "please adjust!" comments: Once you're satisfied.

c. Handling Empty Migration Scripts:

Sometimes, alembic revision --autogenerate might produce an "empty" migration script (with just pass in upgrade() and downgrade()) even if you haven't made functional changes to your models. This can happen due to:

  • Minor differences in how SQLAlchemy reflects types from the database versus your model definitions.
  • Default naming of constraints or indexes if not explicitly set via naming conventions.
  • SQLite specific behaviors if batch mode isn't enabled.

If you get an empty script and you're sure no actual schema change is needed:

  1. Review it: Confirm it's truly empty (only pass or comments).
  2. Delete it: You can safely delete the empty .py script file from alembic/versions/ to keep your history clean.
  3. Proactive Measures: Using naming conventions and render_as_batch=True (as described in section 3.d) helps minimize the occurrence of these empty scripts.

d. Apply the New Migration:

Once you've reviewed (and corrected, if necessary, or deleted if empty and unneeded) the new migration script, apply it to your database if it contains actual changes:

alembic upgrade head
Enter fullscreen mode Exit fullscreen mode

This will execute the upgrade() function in your new migration script.

6. Applying an Existing Migration (After Initial Setup):

If you have already run alembic init and configured Alembic, and you have migration scripts (e.g., from version control), you apply them using:

alembic upgrade head
Enter fullscreen mode Exit fullscreen mode

This command runs the upgrade() function in all unapplied migration scripts.

7. Alembic Modes: Online vs. Offline Migrations

Alembic can operate in two main modes, defined by the run_migrations_online() and run_migrations_offline() functions in alembic/env.py.

Online Migrations (run_migrations_online()):

  • Connects to a live database: Alembic establishes an active connection to your target database.
  • Applies changes directly: When you run alembic upgrade head, Alembic uses this live connection to execute DDL statements (e.g., CREATE TABLE, ALTER TABLE) directly against the database.
  • Most common mode: Used for typical development and deployment where Alembic manages schema changes directly.

Offline Migrations (run_migrations_offline()):

  • Generates SQL scripts: This mode does not require a live database connection to generate the DDL. Instead, it outputs the SQL statements needed to perform the migration.
  • Manual application: The generated SQL scripts can then be reviewed and applied manually by a Database Administrator (DBA) or through other database deployment tools.
  • Use cases: Useful in environments with strict control over database changes, or where developers don't have direct DDL privileges on production databases.

Generating SQL Scripts for Offline Mode:

To generate SQL scripts instead of applying migrations directly, use the --sql flag with alembic upgrade (or downgrade):

alembic upgrade head --sql
Enter fullscreen mode Exit fullscreen mode

This will print the SQL DDL statements to the console. You can redirect this to a file:

alembic upgrade head --sql > migration_to_head.sql
Enter fullscreen mode Exit fullscreen mode

This migration_to_head.sql file can then be used for manual review and application. Alembic still uses the sqlalchemy.url to determine the correct SQL dialect for the script.

8. Understanding Alembic Conceptually and its CLI

To use Alembic effectively, it's helpful to understand its core concepts and common command-line interface (CLI) commands.

Core Concepts:

  • Migration Scripts: These are Python files in the alembic/versions/ directory. Each script represents a single, ordered change to your database schema. They contain an upgrade() function (to apply the change) and a downgrade() function (to revert the change).
  • Revisions: Each migration script has a unique revision ID (a short hash). Alembic uses these IDs to track the order of migrations and the current version of your database schema.
  • Head: The head refers to the latest revision (the most recent migration script) in your project.
  • Base: The base usually refers to an empty schema (before any migrations are applied).
  • alembic_version table: Alembic creates and maintains a table named alembic_version in your database. This table stores the revision ID of the currently applied migration, allowing Alembic to know the database's schema version.

Common CLI Commands:

  • alembic init <directory>: Initializes a new Alembic migration environment in the specified directory (e.g., alembic init alembic).
  • alembic revision -m "<message>": Creates a new, blank migration script with the given message. You would then manually edit this script to define schema changes.
  • alembic revision -m "<message>" --autogenerate: Attempts to automatically generate a migration script by comparing your current SQLAlchemy models (defined by target_metadata in env.py) against the current state of the database. Always review autogenerated scripts.
  • alembic upgrade <revision>: Applies migrations to upgrade the database schema.
  • alembic upgrade head: Upgrades to the latest revision.
  • alembic upgrade +1: Applies the next unapplied migration.
  • alembic upgrade <revision_hash>: Upgrades to a specific revision.
  • alembic downgrade <revision>: Reverts migrations to downgrade the database schema.
  • alembic downgrade base: Downgrades to an empty schema (reverts all migrations).
  • alembic downgrade -1: Reverts the last applied migration.
  • alembic downgrade <revision_hash>: Downgrades to the state before the specified revision was applied.
  • alembic history: Shows the list of revisions, indicating which ones are applied and which are pending.
  • alembic current: Shows the current revision ID of the database.
  • alembic show <revision>: Displays information about a specific revision, including its content.
  • alembic stamp <revision>: Sets the alembic_version table to a specific revision ID without actually running the migration scripts. This is useful for aligning an existing database's schema version with Alembic if migrations were applied manually or if you're initializing Alembic on an existing database.

9. Safely Deleting Alembic Revisions

Deleting Alembic revision scripts requires caution, as it can affect your migration history. The safety depends on whether the revision has been applied.

Scenario 1: Revision NOT applied to any important database (or only to a local dev DB you can reset).

  • Delete the script file: Remove the .py file from alembic/versions/.
  • Adjust down_revision in the next script (if any): If the deleted script was not the "head" (latest), the script that followed it will have its down_revision pointing to the deleted one. Edit this subsequent script to point its down_revision to the revision before the one you deleted.

Example: Scripts A (rev:aaa) -> B (rev:bbb, down_rev:aaa) -> C (rev:ccc, down_rev:bbb). If you delete B, edit C so down_revision = 'aaa'.

Scenario 2: Revision applied to local/dev DB, but NOT to shared/staging/production.

  • Downgrade the database: Revert the changes from your local DB. E.g., if deleting revision bbb which follows aaa, run alembic downgrade aaa.
  • Delete the script file: As in Scenario 1.
  • Adjust down_revision in the next script (if any): As in Scenario 1.

Special Case: Deleted Head Revision After Applying to Local DB:

If you deleted the script for the latest (head) revision after it was applied to your local DB:

  • Downgrade your local database: Run alembic downgrade -1 (or alembic downgrade <revision_id_before_the_deleted_one>). This attempts to run the downgrade() function of the (now missing) script.
  • If downgrade fails due to missing script: For a local dev DB, you might need to:
    • Manually revert the schema changes in the DB.
    • Restore the DB from a backup if available.
    • Or, as a last resort for a dev DB, drop the DB, run alembic upgrade head (to the new head), and repopulate data.
  • After manual changes or restore, you might use alembic stamp <correct_previous_revision_id> to align Alembic's history.
  • Confirm script deletion: Ensure the .py file is gone from alembic/versions/.
  • No down_revision adjustment is needed for other scripts since it was the head.

Scenario 3: Revision applied to shared, staging, or PRODUCTION databases.

  • DO NOT simply delete the script. This will break the migration history for those databases.
  • Instead, create a new "reverting" migration:
    • Generate a new blank script: alembic revision -m "revert_changes_from_revision_xyz"
    • In its upgrade() function, write Alembic operations to undo what the problematic revision did.
    • Apply this new reverting migration: alembic upgrade head.
    • This keeps the history consistent.

General Safety Tips for Deleting/Managing Revisions:

  • Version Control (Git): Essential for your Alembic environment.
  • Test Downgrades: Ensure your downgrade() functions work.
  • Backup Production: Always back up before production migrations.

10. Handling Divergent Migration Histories (Branching and Merging)

In team environments using version control (like Git), it's possible for different developers to independently create migration scripts based on the same parent revision. This leads to "multiple heads" or "branches" in the Alembic migration history.

Identifying Multiple Heads: 

alembic heads
Enter fullscreen mode Exit fullscreen mode

This command will list the revision IDs of all current branch tips.

The alembic merge Command:

The alembic merge command creates a new, special migration script that joins these divergent branches back into a single, linear history.

alembic merge <revision1> <revision2> ... -m "Descriptive merge message"
Enter fullscreen mode Exit fullscreen mode

Or, to merge all current heads automatically:

alembic merge heads -m "Merge all current heads"
Enter fullscreen mode Exit fullscreen mode

The generated merge script typically has empty upgrade() and downgrade() functions. Its key feature is that its down_revision attribute will be a tuple containing the revision IDs of the heads it merged, thus linking the branches.

Resolving Conflicts:

alembic merge does not resolve schema conflicts. If the divergent branches made conflicting changes to the schema (e.g., altering the same column differently), you must resolve these conflicts manually. This might involve:

  • Editing one or more of the branch migration scripts before running alembic merge.
  • Creating a new migration script after the merge script to apply the necessary corrections.

Careful review and communication within the team are crucial when dealing with branched migrations.

Applying the Merge Script:

After the merge script is generated, run alembic upgrade head to apply it and update the alembic_version table. This will consolidate your history to a single head.

11. Summary of Alembic Workflow for Schema Changes:

  • Modify your SQLModel model(s).
  • Generate a new migration: alembic revision -m "description" --autogenerate
  • Always review and, if necessary, correct the generated migration script.
  • Apply the migration: alembic upgrade head (or generate SQL using --sql for offline application).

This concludes the basic introduction to SQLModel and Alembic for defining models, interacting with data, and managing database schema migrations!

Complete Example Code (main.py)

This main.py file would contain your SQLModel definitions and the CRUD operation functions. Note that for a project using Alembic, the create_db_and_tables() function would typically only be used for initial local development database setup if the database doesn't exist at all, or not used in favor of alembic upgrade head.

import uuid
from typing import Optional

from sqlmodel import Field, SQLModel, create_engine, Session, select 
from sqlalchemy import Column, text, func  # text & func for potential advanced examples
from sqlalchemy.types import JSON
from sqlalchemy.orm.attributes import flag_modified  # Import flag_modified
from pydantic import ConfigDict 
from datetime import datetime  # For potential advanced examples

# --- Part 2: Defining Your First SQLModel ---
# This is the FINAL version of the Hero model, after the 'allegiance' field is added via migration.
class Hero(SQLModel, table=True):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True, index=True, nullable=False)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)
    powers: Optional[list[str]] = Field(default=None, sa_column=Column(JSON))
    attributes_data: Optional[dict[str, any]] = Field(default=None, sa_column=Column(JSON))
    allegiance: Optional[str] = Field(default=None, index=True)  # New field, added via migration

# --- Part 3: Creating the Database and Tables ---
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)  # echo=True logs SQL statements

def create_db_and_tables():
    # This function is mainly for initial setup if not using Alembic from the very first creation
    # or for tests where you want a fresh DB.
    # In a production-like workflow, Alembic handles table creation and updates.
    SQLModel.metadata.create_all(engine)

# --- Part 4: Interacting with Data (CRUD Operations) ---
# These CRUD examples now include the 'allegiance' field.
def create_heroes_example():
    print("\n--- Creating heroes ---")
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", powers=["healing", "teleportation"], attributes_data={"suit_color": "red", "humor_level": 10}, allegiance="Mercs for Money")
    hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador", age=16, powers=["web-slinging", "wall-crawling"], attributes_data={"mentor": "Deadpond", "school": "Midtown High"}, allegiance="Web Warriors")
    hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48, powers=["genius intellect", "powered armor"], attributes_data={"company": "Sharp Industries", "armor_mark": "XLII"}, allegiance="Avengers")

    with Session(engine) as session:
        session.add(hero_1)
        session.add(hero_2)
        session.add(hero_3)
        session.commit()

        # Refresh to get IDs and other DB-generated data
        session.refresh(hero_1)
        session.refresh(hero_2)
        session.refresh(hero_3)
        print(f"Created Hero: ID={hero_1.id}, Name={hero_1.name}, Allegiance={hero_1.allegiance}")
        print(f"Created Hero: ID={hero_2.id}, Name={hero_2.name}, Allegiance={hero_2.allegiance}")
        print(f"Created Hero: ID={hero_3.id}, Name={hero_3.name}, Allegiance={hero_3.allegiance}")

def read_heroes_example():
    print("\n--- Reading heroes ---")
    with Session(engine) as session:
        statement = select(Hero)
        heroes = session.exec(statement).all()
        print("All heroes in database:")
        for hero in heroes:
            print(f"  ID: {hero.id}, Name: {hero.name}, Age: {hero.age}, Powers: {hero.powers}, Attributes: {hero.attributes_data}, Allegiance: {hero.allegiance}")

        print("\nReading heroes older than 30:")
        statement_older = select(Hero).where(Hero.age > 30)
        heroes_older = session.exec(statement_older).all()
        for hero in heroes_older:
            print(f"  Name: {hero.name}, Age: {hero.age}, Allegiance: {hero.allegiance}")

        print("\nReading hero 'Deadpond':")
        statement_deadpond = select(Hero).where(Hero.name == "Deadpond")
        deadpond = session.exec(statement_deadpond).first()
        if deadpond:
            print(f"  Found: {deadpond.name}, Secret Name: {deadpond.secret_name}, Allegiance: {deadpond.allegiance}")
        else:
            print("  Deadpond not found.")

def update_heroes_example():
    print("\n--- Updating heroes ---")
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Spider-Boy")
        spider_boy = session.exec(statement).first()

        if spider_boy:
            print(f"Before update - Spider-Boy: Age={spider_boy.age}, Powers={spider_boy.powers}, Allegiance={spider_boy.allegiance}")
            spider_boy.age = 17 
            spider_boy.allegiance = "Champions"  # Update allegiance
            if spider_boy.powers: 
                 spider_boy.powers.append("super-strength (mild)")
            else:
                 spider_boy.powers = ["super-strength (mild)"]

            flag_modified(spider_boy, "powers")  # Ensure modification to list is tracked

            session.add(spider_boy) 
            session.commit() 
            session.refresh(spider_boy)
            print(f"After update - Spider-Boy: Age={spider_boy.age}, Powers={spider_boy.powers}, Allegiance={spider_boy.allegiance}")
        else:
            print("Spider-Boy not found for update.")

def delete_heroes_example():
    print("\n--- Deleting heroes ---")
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name == "Rusty-Man")
        rusty_man = session.exec(statement).first()

        if rusty_man:
            print(f"Deleting hero: {rusty_man.name}")
            session.delete(rusty_man)
            session.commit()
            print(f"{rusty_man.name} deleted.")

            verify_statement = select(Hero).where(Hero.name == "Rusty-Man")
            deleted_hero = session.exec(verify_statement).first()
            if not deleted_hero:
                print(f"Verified: {rusty_man.name} is no longer in the database.")
            else:
                print(f"Verification FAILED: {rusty_man.name} still found.")
        else:
            print("Rusty-Man not found for deletion.")

# Main execution block (for demonstration)
# In a real application, you might not run all these sequentially like this.
# Alembic would be used for schema setup and migrations.
if __name__ == "__main__":
    # For initial setup if DB doesn't exist and you're not using Alembic for the very first creation:
    # create_db_and_tables() 
    # print("Database and tables created (if they didn't exist by create_db_and_tables).")
    # print("For subsequent schema changes, use Alembic migrations.")

    # The following CRUD operations assume the tables exist (e.g., created by Alembic or create_db_and_tables)
    # If running with Alembic, ensure you've run `alembic upgrade head` first.

    print("Running CRUD examples... (Ensure database schema is up-to-date with Alembic)")
    create_heroes_example()
    read_heroes_example()
    update_heroes_example()
    delete_heroes_example()
    print("\n--- Reading heroes after all operations ---")
    read_heroes_example()  # Read again to see the final state
Enter fullscreen mode Exit fullscreen mode

Top comments (0)