Database¶

Model comes with built-in database layers for SQLite, MariaDB, and MySQL, which you can use directly to read and modify data, run SQL, manage tables, and work with database connections without defining a model.

Connect to a database¶

Database objects connect when they are created.

SQLite¶

With no arguments, SQLite creates an in-memory database. Pass a file path if you want the data to persist:

from model.database import SQLiteDatabase

db = SQLiteDatabase()

file_db = SQLiteDatabase("./data/app.db")

An in-memory database belongs to one connection. Another thread gets a separate, empty database, so use a file if multiple threads need to share the same data.

You can also pass timeout, check_same_thread, and cached_statements through other_config:

db = SQLiteDatabase(
    "./data/app.db",
    other_config={
        "timeout": 10,
    },
)

These options behave the same as they do with sqlite3.connect.

MariaDB and MySQL¶

import os

from model.database import MySQLDatabase

db = MySQLDatabase(
    host="127.0.0.1",
    port=3306,
    user="app",
    password=os.environ["DB_PASSWORD"],
    database="world",
)

The main connection options are:

• host
• port
• user
• password
• database
• other_config

Use other_config for additional PyMySQL settings such as timeouts, charset, TLS, client flags, or a Unix socket.

Unknown options raise DatabaseError.

MySQL connection pool¶

Use MySQLDatabasePool when concurrent work should share a limited number of connections:

from model.database import MySQLDatabasePool

db = MySQLDatabasePool(
    pool_size=8,
    pool_timeout=5,
    host="127.0.0.1",
    user="app",
    password="secret",
    database="world",
)

pool_size must be between 1 and 32.

pool_timeout controls how long an operation waits for a free connection. Set it to 0 to fail immediately when none is available.

Each query or built-in database method borrows one connection and returns it when finished.

The pool accepts the normal MySQL connection options, plus optionally fallback_servers: an ordered list of connection-option dictionaries containing fallback values such as an alternate host or port.

Read and change data¶

The examples below use SQLite's ? placeholders.

With MariaDB and MySQL, use %s instead.

Find rows¶

Use find_one() when you want one matching row:

city = db.find_one(
    "city",
    "Name = ?",
    ["Tirana"],
    columns=["ID", "Name", "Population"],
)

You do not need to include WHERE.

Use find_all() for multiple rows:

cities = db.find_all(
    "city",
    "CountryCode = ? ORDER BY Population DESC",
    ["ALB"],
    columns=["Name", "Population"],
)

You can also add a LIMIT clause to limit the total results returned.

Use count() when you only need the number of matching rows:

large_city_count = db.count(
    "city",
    "Population >= ?",
    [100_000],
)

When find_one() could match several rows, add ORDER BY if you need a predictable result.

Insert rows¶

Use insert() to add one row:

result = db.insert(
    "city",
    {
        "Name": "Saranda",
        "CountryCode": "ALB",
        "Population": 50_000,
    },
)

print(result.rowcount)
print(result.lastrowid)

It returns an InsertResult.

Call insert() with only the table name to insert a row using database defaults and any auto-incrementing key.

Use insert_many() to add several rows with one statement:

rows = [
    {
        "Name": "Tirana",
        "CountryCode": "ALB",
        "Population": 418_495,
    },
    {
        "Name": "Saranda",
        "CountryCode": "ALB",
        "Population": 50_000,
    },
]

result = db.insert_many("city", rows)

The list must not be empty, and every dictionary must have the same keys in the same order.

For insert_many(), lastrowid is only the ID reported for the statement. It does not identify every inserted row.

Update rows¶

Use update() to change matching rows:

changed = db.update(
    "city",
    {"Population": 420_000},
    "Name = ?",
    ["Tirana"],
)

It returns the number of affected rows.

Insert or update¶

Use insert_or_update() when you want to insert a row if it does not exist, or update it when a matching unique value already exists:

result = db.insert_or_update(
    "user",
    {
        "email": "ada@example.com",
        "display_name": "Ada",
    },
    skip_update_on_columns=["email"],
)

print(result.is_insert)
print(result.is_update)
print(result.lastrowid)

It returns an InsertOrUpdateResult.

Delete rows¶

Use delete() to remove matching rows:

deleted = db.delete(
    "city",
    "CountryCode = ?",
    ["ALB"],
)

It returns the number of affected rows.

Run SQL directly¶

Use query() when you need SQL that does not fit the built-in methods, such as joins, expressions, aliases, or more complex statements.

with db.query(
    "SELECT Name, Population FROM city WHERE CountryCode = ?",
    ["ALB"],
) as cursor:
    cities = cursor.fetchall()

Keep reads and other cursor work inside the with block. The cursor closes when the block ends.

When using a connection pool, its connection is also returned to the pool.

Common cursor features include:

• fetchone()
• fetchall()
• rowcount
• lastrowid

Returned rows¶

Rows support both dictionary-style and attribute access:

with db.query(
    "SELECT Name, CountryCode FROM city LIMIT 1"
) as cursor:
    city = cursor.fetchone()

if city is not None:
    print(city["Name"])
    print(city.Name)

You can also use get():

print(city.get("countrycode"))

get() is case-insensitive.

Use dict(row) when you want a regular dictionary.

Transactions¶

Database objects use autocommit by default.

A query() block controls how long its cursor stays open. It does not automatically start, commit, or roll back a transaction.

The most portable approach is to keep the entire transaction on one cursor:

with db.query("BEGIN") as cursor:
    try:
        cursor.execute(
            "UPDATE account SET balance = balance - ? WHERE id = ?",
            [20, 1],
        )

        cursor.execute(
            "UPDATE account SET balance = balance + ? WHERE id = ?",
            [20, 2],
        )

    except BaseException:
        cursor.execute("ROLLBACK")
        raise

    else:
        cursor.execute("COMMIT")

Use %s placeholders with MySQL.

Manage tables and columns¶

Use these methods to inspect and change tables, including migrations inside pre_sync and post_sync model hooks. See Sync and migrations for more.

Check whether a table exists:

db.has_table("city")

Check whether a column exists:

db.has_column(
    table="city",
    column="CountryCode",
)

You can optionally check its type:

db.has_column(
    table="city",
    column="CountryCode",
    type="CHAR",
)

Column names are compared without caring about letter case.

Type checks also ignore case and match the beginning of the type, so CHAR matches CHAR(3).

Rename a column:

db.rename_column(
    table="city",
    old_column="CountryCode",
    new_column="country_code",
)

Remove a column:

db.drop_column(
    table="city",
    column="obsolete",
)

Rename a table:

db.rename_table(
    old_table="city",
    new_table="cities",
)

Remove a table:

db.drop_table("cities")

When SQLite cannot directly remove a constrained column, Model tries to rebuild the table while preserving the rest of its structure.

If that cannot be done safely, it raises DatabaseError.