simple-sqlite3-orm

Simple Python SQLite3 ORM

A simple yet powerful SQLite3 ORM based on Python's sqlite3 DB engine, powered by pydantic.

It targets basic CRUD operations and does it well, while also opened to complicated use cases.

Features and hightlights

• Light-weight sqlite3 ORM based on Python3's std sqlite3 DB engine, with only pydantic and typing_extensions as dependencies.
• Support defining your database and table as code.
• Provides simple and clean APIs for basic CRUD operatations.
• All functions and types are fully typed and docstrings applied.
• Opened to advanced and more complicated use cases with helper functions, extra APIs and sqlite3 specific constants.

Natively supported Python types

Besides the sqlite3 natively supported python types,simple-sqlite3-orm also adds direct support to the following python types:

• Enums types: IntEnum and StrEnum.
• Literal types: str Literal and int Literal.
• Supported types that wrapped within Optional(like Optional[str]).

simple-sqlite3-orm also datetime support with the following types:

• DatetimeUnixTimestamp: will be serialized and stored as REAL in database.
• DatetimeUnixTimestampInt: will be serialized and stored as INTEGER in database.
• DatetimeISO8601: will be serialized into ISO8601 format string and stored as TEXT in database.

Installation

pip install simple-sqlite3-orm

simple-sqlite3-orm supports Python 3.8+.

Basic usage

This chapter only shows very basic(thus simple) usage of CRUD operations, there are also many extra APIs available for advanced use cases.

simple-sqlite3-orm applies docstrings to most of the APIs, you can always refer to docstrings for help and more information.

Define your table as code

simple-sqlite3-orm provides TableSpec as base for you to define table.

TableSpec subclasses pydantic's BaseModel, so you can follow your experience of using pydantic to define your table as code.

For a more complicated example, see sample_db.

from typing import Literal
from simple_sqlite3_orm import ConstrainRepr, TableSpec, TypeAffinityRepr

# ------ Table definition ------ #

class MyTable(TableSpec):
    entry_id: Annotated[int, ConstrainRepr("PRIMARY KEY")]
    entry_type: Annotated[
        Literal["A", "B", "C"],
        ConstrainRepr("NOT NULL", ("CHECK", "entry_type IN (A,B,C)"))
    ]
    entry_token: bytes

    # A custom type that defines serializer/deserializer in pydantic way,
    #   this custom type is serialized into bytes and stored as BLOB in database.
    special_attrs: Annotated[SpecialAttrsType, TypeAffinityRepr(bytes), ConstrainRepr("NOT NULL")]

(Recommended) Define typing helpers for your table

It is recommended to define two typing helpers for your table:

• A TypedDict for generating col/value mapping with type check.
• A ColsSelector for generating a tuple for columns name selection with type check and completion suggestions.

simple_sqlite3_orm APIs take mappings as params, and take tuples for columns selections. You can utilize the defined typing helpers to enable static type chec with completion suggestions over column names when using APIs, preventing column name typos or unsynced updates to column names.

from typing import TypedDict
from simple_sqlite3_orm import ColsSelectFactory

# ------------ TypedDict ------------ #

# NOTE: `total` param below allows only specifying some col/value pairs
class MyTableCols(TypedDict, total=False):
    # no need to copy and paste the full type annotations from the actual TableSpec,
    #   only the actual type is needed.
    entry_id: int
    entry_type: Literal["A", "B", "C"]
    entry_token: bytes
    special_attrs: SpecialAttrsType

# examples:

good_rows = MyTableCols(entry_id=123)
bad_rows_type_invalid = MyTableCols(entry_id="not_a_int") # type check error
bad_rows_unknown_col = MyTableCols(unknown="unknown") # type check error

# ------------ ColsSelector ------------ #

MyTableColsSelector = ColsSelectFactory[Literal["entry_id", "entry_type", "entry_token", "special_attrs"]]

# examples:

good_cols_selection = MyTableColsSelector("entry_id", "special_attrs")
bad_cols_selection = MyTableColsSelector("entyr_di") # type check error

Define your database as code

After the table definition is ready, you can further define ORM types.

simple-sqlite3-orm provides ORMBase for you to define the ORM with table you defined previously. ORMBase supports defining database as code with specifying table_name, table create configuration and indexes for deterministically bootstrapping new empty database file.

from simple_sqlite3_orm import CreateIndexParams, CreateTableParams, ORMBase

class MyORM(ORMBase[MyTable]):

    orm_bootstrap_table_name = "my_table"
    orm_bootstrap_create_table_params = CreateTableParams(without_rowid=True)
    orm_bootstrap_indexes_params = [
        CreateIndexParams(index_name="entry_token_index", index_cols=MyTableColsSelector("entry_token"))
    ]

Bootstrap new database

After defining the ORM, you can bootstrap a new empty database, create table(and indexes) deterministically as follow:

import sqlite3

conn = sqlite3.connect("my_db.sqlite3")
orm = MyORM(conn)

orm.orm_bootstrap_db()

Alternatively, you can also use orm_create_table and orm_create_index separately to bootstrap a new database.

Insert rows

You can use orm_insert_entry or orm_insert_mapping to insert exactly one entry:

entry_to_insert: MyTable
mapping_to_insert: MyTableCols

# insert a row by MyTable instance
orm.orm_insert_entry(entry_to_insert)

# insert a row by mapping as MyTableCols TypedDict
#   with a mapping, you can insert partially set row and let DB engine fill
#   the unprovided cols with DEFAULT value or NULL.
orm.orm_insert_mapping(mapping_to_insert)

Or you can insert a bunch of entries by an iterable of entries:

entries_to_insert: Iterable[MyTable]
mappings_to_insert: Iterable[MyTableCols]

inserted_entries_count = orm.orm_insert_entries(entries_to_insert)
inserted_entries_count = orm.orm_insert_mappings(mappings_to_insert)

Select rows

You can select entries by matching column(s) from database:

res_gen: Generator[MyTable] = orm.orm_select_entries(MyTableCols(entry_type="A", entry_token=b"abcdef"))

# process each selected entry here
for entry in res_gen: ...

Update rows

You can update specific rows with one set of params as follow:

# specify rows by matching cols
#   WHERE stmt will be generated from `where_cols_value`.
orm.orm_update_entries(
    set_values=MyTableCols(entry_token="ccddee123", entry_type="C"),
    where_cols_value=MyTableCols(entry_id=123),
)

# alteratively, you can directly provide the WHERE stmt and `extra_params` for the query execution.
#   be careful to not use the columns's named-placeholder used by `set_values`.
orm.orm_update_entries(
    set_values=MyTableCols(entry_token="ccddee123", entry_type="C"),
    where_stmt="WHERE entry_id > :entry_lower_bound AND entry_id < :entry_upper_bound",
    _extra_params={"entry_lower_bound": 123, "entry_upper_bound": 456}
)

Also, there is an executemany version of ORM update API, orm_update_entries_many, which you can use many sets of params for the same UPDATE query execution. Using this API is SIGNIFICANTLY faster with lower memory usage than calling orm_update_entries each time in a for loop.

set_cols_value_iter: Iterable[MyTableCols]
where_cols_value_iter: Iterable[Mapping[str, Any]]

updated_rows_count: int = orm.orm_update_entries_many(
    set_cols=MyTableColsSelector("entry_id", "entry_token", "entry_type"),
    where_cols=MyTableColsSelector("entry_id"),
    set_cols_value=set_cols_value_iter,
    where_cols_value=where_cols_value_iter,
)

Delete rows

Like select operation, you can detele entries by matching column(s):

affected_row_counts: int = orm.orm_delete_entries(MyTableCols(entry_type="C"))

ORM pool support

simple-sqlite3-orm also provides ORM threadpool(ORMThreadPoolBase) and asyncio ORM(AsyncORMBase, experimental) supports.

ORM threadpool and asyncio ORM implements most of the APIs available in ORMBase, except for the orm_conn API.

License

simple-sqlite3-orm is licensed under Apache 2.0 License.