Metadata-Version: 2.4
Name: sqloquent
Version: 0.7.3
Summary: SQL/ORM library with included bindings for sqlite. Inspired by Laravel and in particular Eloquent, but built with a distributed use case in mind (e.g. Merkle DAGs).
Project-URL: Homepage, https://github.com/k98kurz/sqloquent
Project-URL: Repository, https://github.com/k98kurz/sqloquent
Project-URL: Bug Tracker, https://github.com/k98kurz/sqloquent/issues
Author-email: k98kurz <k98kurz@gmail.com>
License: Copyright (c) 2026 Jonathan Voss (k98kurz)
        
        Permission to use, copy, modify, and/or distribute this software
        for any purpose with or without fee is hereby granted, provided
        that the above copyright notice and this permission notice appear in
        all copies.
        
        THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL
        WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED
        WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE
        AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR
        CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
        OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
        NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
        CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
Classifier: Development Status :: 4 - Beta
Classifier: License :: OSI Approved :: ISC License (ISCL)
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Database :: Front-Ends
Requires-Python: >=3.10
Requires-Dist: packify>=0.3.1
Provides-Extra: asyncql
Requires-Dist: aiosqlite>=0.19.0; extra == 'asyncql'
Requires-Dist: nest-asyncio==1.6.0; extra == 'asyncql'
Description-Content-Type: text/markdown

# Sqloquent

This is a SQL library with included bindings for sqlite. Inspired by Laravel and
in particular Eloquent.

## Overview

This package provides a set of interfaces and classes to make using a SQL
database easier and simpler, both synchronously and by using asyncio. (See
section below for full list.)

The primary features are the `SqlQueryBuilder`, `SqlModel`, and `HashedModel` base
classes (or `AsyncSqlQueryBuilder`, `AsyncSqlModel`, and `AsyncHashedModel` for use
with asyncio) and ORM system. The `SqlQueryBuilder` uses a builder pattern to build
and execute a query from various clauses. The `SqlModel` handles encoding,
persisting, reading, and decoding models that correspond to rows. The query builder
can be used without a model, in which case a dynamic model will be created. Any
grouping will result in `get` returning `Row`s, and joining will result in `get`
returning `JoinedModel`s.

Below are some examples of using the query builder without models:

```python
from sqloquent import SqlQueryBuilder

query = lambda table, columns: SqlQueryBuilder(
    table=table, columns=columns, connection_info='temp.db'
)

# count the number of matches
sqb = query('some_table', ['id', 'etc']).join(
    'some_other_table',
    on=['id', 'some_id'],
    joined_table_columns=['id', 'some_id', 'data']
)
count = sqb.count()

# chunk through them 1000 at a time
for chunk in sqb.chunk(1000):
    for joined_model in chunk:
        ...

# or just get them all
results = sqb.get()

# or use a condition
results = query('some_table', ['id', 'etc']).where(contains={'etc': 'something'}).get()
# or equivalently
results = query('some_table', ['id', 'etc']).contains(etc='something').get()
```

Or for asyncio:

```python
from asyncio import run
from sqloquent.asyncql import AsyncSqlQueryBuilder

query = lambda table, columns: AsyncSqlQueryBuilder(
    table=table, columns=columns, connection_info='temp.db'
)

sqb = query('some_table', ['id', 'etc']).join(
    'some_other_table',
    on=['id', 'some_id'],
    joined_table_columns=['id', 'some_id', 'data']
)

# count the number of matches
count = run(sqb.count())

# chunk through them 1000 at a time
async def chunk_it(sqb):
    async for chunk in sqb.chunk(1000):
        for joined_model in chunk:
            ...
run(chunk_it(sqb))

# or just get them all
results = run(sqb.get())

# or use a condition
results = run(query('some_table', ['id', 'etc']).where(contains={'etc': 'something'}).get())
# or equivalently
results = run(query('some_table', ['id', 'etc']).contains(etc='something').get())
```

The base classes have a default binding to sqlite3 via the `SqliteContext`
class, but they can be coupled to any SQL database client. See the "Usage"
section below for detailed instructions for the latter.

Additionally, three classes, `DeletedModel`, `HashedModel`, and `Attachment`
have been supplied to allow easy implementation of a system that includes a
cryptographic audit trail. `DeletedModel` corresponds to any deleted model from
a class that extends `HashedModel` and includes a `restore` method that can
restore the deleted record.

There is an included CLI tool that generates code scaffolding for models and
migrations, as well as track, apply, rollback, and refresh migrations. See the
[dox.md](https://github.com/k98kurz/sqloquent/blob/v0.7.3/docs/dox.md) and
[asyncql_dox.md](https://github.com/k98kurz/sqloquent/blob/v0.7.3/docs/asyncql_dox.md)
files generated by [autodox](https://pypi.org/project/autodox) for full
documentation, or read
[interfaces.md](https://github.com/k98kurz/sqloquent/blob/v0.7.3/docs/interfaces.md)
and
[async_interfaces.md](https://github.com/k98kurz/sqloquent/blob/v0.7.3/docs/async_interfaces.md)
for documentation on just the interfaces or
[tools.md](https://github.com/k98kurz/sqloquent/blob/v0.7.3/docs/tools.md) for
information about the bundled tools.

## Status

See open issues that are planned for a future release
[here](https://github.com/k98kurz/sqloquent/issues?q=is%3Aissue+is%3Aopen+label%3Aplanned).
Historic changes are summarized in the
[changelog](https://github.com/k98kurz/sqloquent/blob/master/changelog.md).

Currently, only the basic sqlite3 types (affinities) of text, blob, integer,
real, numeric, and boolean are supported by the migration system. Support for
all data types is planned for a future release:
[issue #8](https://github.com/k98kurz/sqloquent/issues/8).

## Setup and Usage

Requires Python 3.10+. This has not been tested with older Python versions.

### Setup

```bash
pip install sqloquent
```

To use the async version, instead install with the following:

```bash
pip install sqloquent[asyncql]
```

### Quick Start

Getting started is pretty quick using the sqloquent CLI:

```bash
export CONNECTION_STRING='database.db' MAKE_WITH_CONNSTRING=1
mkdir -p migrations
mkdir -p models
sqloquent make model GraphNode --columns "id=str,details=bytes,parent_ids=str" > models/GraphNode.py
cat <<EOF >> models/__init__.py
from .GraphNode import GraphNode
from sqloquent import contains, within

GraphNode.parents = contains(GraphNode, GraphNode, 'parent_ids')
GraphNode.children = within(GraphNode, GraphNode, 'parent_ids')
__all__ = [
    GraphNode,
]
EOF

sqloquent make migration --model GraphNode models/GraphNode.py > migrations/001_create_graph_nodes.py
sqloquent automigrate migrations/
```

Then, the model can be used within an app or script:

```python
from models import GraphNode

print(GraphNode.query().count())
parent = GraphNode.insert({'details': b'parent: 1234'})
child = GraphNode.insert({'details': b'child: 4321', 'parent_ids': parent.id})
print(GraphNode.query().count())
print([c.id for c in parent.children], child.id)
print([p.id for p in child.parents], parent.id)
```

### Usage

There are two primary ways to use this package: either with a bundled sqlite3
coupling or with a custom coupling to an arbitrary SQL database client. The
cryptographic audit trail features can be used with any SQL database coupling.

Note that if you create a custom async DB coupling, you will also need to create
a non-async coupling to use the migration system. Also note that at the moment,
this library has only been tested with sqlite3.

#### CLI Tool

For ease of development, a CLI tool is included which can be used for generating
code scaffolds/boilerplates and for managing migrations. After installing via
pip, run `sqloquent` in the terminal to view the help text.

The CLI tool can generate models and migrations, including the ability to
generate migrations from completed models. Migrations can be handled manually or
using an automatic method that tracks migrations via a `migrations` table. To
use the migration tools, the environment variable `CONNECTION_STRING` should be
set either in the CLI environment or in a .env file, e.g.
`CONNECTION_STRING=path/to/file.db`. To insert this connection string into
generated scaffold code, also define a `MAKE_WITH_CONNSTRING` environment
variable and set it to anything other than "false" or "0"; this is a convenience
feature for working with sqlite3, since that is the only bundled coupling, but
overwriting the `connection_info` attribute on models at the app execution entry
point is probably a better strategy -- if using another SQL binding, the
connection info should be injected into the context manager (see section about
binding to other SQL databases/engines below).

Additionally, the functionality exposed by the CLI tool can be accessed
programmatically through `sqloquent.tools`.

#### Connection Information

Connection information can be bound or injected in several places:

- Bound to each individual model
- Injected into the query builder
- Bound to the query builder
- Bound to the db context manager

Items higher in the list will override those lower in the list. For example, if
you set the `connection_info` attribute on a model class or instance, it will be
used for interactions with the db originating from that model class or instance,
respectively. If you set the `connection_info` attribute on the query builder
class, it will be used, but if you pass it as a parameter to initialize a query
builder, that paramter will be used instead.

#### Connection Pooling

In applications that involve high database throughput, it is possible to use
connection pooling to improve performance/reduce IO overhead. The syntax is as
follows:

```python
import SomeSqlModel # assuming the connection_info is set on the model
from sqloquent import SqliteContext

def main():
    with SqliteContext(SomeSqlModel.connection_info) as cursor:
        ... # application logic goes here
```

#### Examples

The most thorough examples are the integration tests. The model files for the
first can be found
[here](https://github.com/k98kurz/sqloquent/tree/master/tests/integration_vectors/models),
and the test itself is
[here](https://github.com/k98kurz/sqloquent/blob/master/tests/test_integration.py#L73).

The async versions can be found here:
- [models](https://github.com/k98kurz/sqloquent/tree/master/tests/integration_vectors/asyncmodels)
- [test](https://github.com/k98kurz/sqloquent/blob/master/tests/test_async_integration.py#L74)

The models were scaffolded using the CLI tool, then the details filled out in
each. The relations were set up in the `__init__.py` file. The integration test
generates migrations from these classes using the CLI tool, automigrates using
the CLI tool, then does inserts/updates/deletes and checks the db for
correctness. (These files provide a basic schema for correspondent banking.)

The second integration test is outlined in the "Using the ORM" section below.

#### Models

Models should extend `SqlModel` or a model that extends `SqlModel` and couples
to another database client. To use the supplied sqlite3 coupling without the
cryptographic features, extend the `SqlModel`, filling these attributes as shown
below:

- `table: str`: the name of the table
- `columns: tuple[str]`: the ordered tuple of column names
- annotations for columns as desired

Additionally, set up any relevant relations using the ORM helper methods.

The CLI tool will produce a scaffold for a model. For example, running
`sqloquent make model Thing --hashed --columns "id,name,stuff=str|None"` will
produce the following:

```python
from sqloquent import HashedModel


class stuff(HashedModel):
    connection_info: str = ''
    table: str = 'things'
    id_column: str = 'id'
    columns: tuple[str] = ('id', 'name', 'stuff')
    id: str
    name: str
    stuff: str|None
```

Specify `--async` to use an async model. For example, running
`sqloquent make model Person --columns id,name --async` will produce the
following:

```python
from sqloquent.asyncql import AsyncSqlModel


class Person(AsyncSqlModel):
    connection_info: str = ''
    table: str = 'persons'
    id_column: str = 'id'
    columns: tuple[str] = ('id', 'name')
    id: str
    name: str
```

Below is a more complex example with relations.

```python
from __future__ import annotations
from sqloquent import SqlModel, has_many, belongs_to, RelatedModel, RelatedCollection
import json

connection_string = ''

with open('.env', 'r') as f:
    lines = f.readlines()
    for l in lines:
        if l[:18] == 'CONNECTION_STRING=':
            connection_string = l[18:-1]

class ModelA(SqlModel):
    connection_info = connection_string
    table: str = 'model_a'
    columns: tuple = ('id', 'name', 'details')
    id: str
    name: str
    details: str
    model_b: RelatedCollection

    @property
    def details(self) -> dict:
        """Decode json str to dict."""
        return json.loads(self.data.get('details', '{}'))

    @details.setter
    def details(self, val: dict) -> None:
        """Sets details and encodes to json str."""
        self.data['details'] = json.dumps(val or '{}')

class ModelB(SqlModel):
    connection_info = connection_string
    table: str = 'model_b'
    columns: tuple = ('id', 'name', 'model_a_id', 'number')
    id: str
    name: str
    model_a_id: str
    number: int
    model_a: RelatedModel


ModelA.model_b = has_many(ModelA, ModelB, 'model_a_id')
ModelB.model_a = belongs_to(ModelB, ModelA, 'model_a_id')


if __name__ == "__main__":
    model_a = ModelA.insert({'name': 'Some ModelA'})
    model_b = ModelB({'name': 'Some ModelB'})
    model_b.details = {'something': 'important(?)'}
    model_b.save()
    assert type(model_a.details) is dict
    assert type(model_b.details) is dict
    model_b.model_a = model_a
    model_b.model_a().save()
    model_a.model_b().reload()
    assert model_a.model_b[0].data['id'] == model_b.id
    assert model_a.model_b[0].id == model_b.id
    ModelA.query().delete()
    ModelB.query().delete()
    print("success")
```

To use this, save the code snippet as "example.py" and run the following to set
up the database and then run the script:

```bash
sqloquent make migration --model ModelA example.py > model_a_migration.py
sqloquent make migration --model ModelB example.py > model_b_migration.py
sqloquent migrate model_a_migration.py
sqloquent migrate model_b_migration.py
python example.py
```

It is noteworthy that every column in the `columns` class attribute will be
made into a property that accesses the underlying data stored in the `data`
dict (the annotation just helps the code editor/LSP pick up on this). This will
not work for any column name that collides with an existing class attribute or
method, and the behavior can be disabled by adding a class attribute called
`disable_column_property_mapping`; all row data will still be accessible via the
`data` attribute on each instance regardless of name collision or feature
disabling.

As of v0.5.2, models will contain a `data_original` dict in addition to the
`data` dict to track changes between database operations, and these will be
visible to most event handlers through the `self` kwarg.

If you do not want to use the bundled ORM system, set up any relevant relations
with `_{related_name}: RelatedModel` attributes and
`{related_name}(self, reload: bool = False)` methods. Dicts should be encoded
using `json.dumps` and stored in text columns. More flexibility can be gained at
the expense of performance by using the packify package, e.g. to encode sets or
classes that implement the `packify.Packable` interface.

```python
from sqloquent import SqlModel


class ModelA(SqlModel):
    table: str = 'model_a'
    columns: tuple = ('id', 'name', 'details')
    id: str
    name: str
    _model_b: ModelB|None = None
    _details: dict|None = None

    def model_b(self, reload: bool = False) -> list[ModelB]:
        """The related model."""
        if self._model_b is None or reload:
            self._model_b = ModelB.query({'model_a_id': self.data['id']}).get()
        return self._model_b

    def set_model_b(self, model_b: ModelB) -> ModelA:
        """Helper method to save some lines."""
        model_b.data['model_a_id'] = self.data['id']
        model_b._model_a = self
        model_b.save()
        self._model_b = model_b
        return self

    @property
    def details(self) -> dict:
        """Decode json str to dict."""
        return json.loads(self.data.get('details', '{}'))

    @details.setter
    def details(self, val: dict) -> None:
        """Sets details and encodes to json str."""
        self.data['details'] = json.dumps(val or '{}')

class ModelB(SqlModel):
    table: str = 'model_b'
    columns: tuple = ('id', 'name', 'model_a_id', 'number')
    id: str
    name: str
    model_a_id: str
    number: int
    _model_a: ModelA|None = None

    def model_a(self, reload: bool = False) -> Optional[ModelA]:
        """Return the related model."""
        if self._model_a is None or reload:
            self._model_a = ModelA.find(self.data['model_a_id'])
        return self._model_a

    def set_model_a(self, model_a: ModelA) -> ModelB:
        """Helper method to save some lines."""
        self.data['model_a_id'] = model_a.data['id']
        self._model_a = model_a
        model_a._model_b = self
        return self.save()
```

#### Using the ORM

The ORM is comprised of 6 classes inheriting from `Relation` and implementing
the `RelationProtocol`: `HasOne`, `HasMany`, `BelongsTo`, `BelongsToMany`,
`Contains`, and `Within`. The async version is equivalent with `Async` prefixes.
Note that currently the async ORM may create `ResourceWarning`s when properties
are accessed.

The recommended way to use the ORM is with the helper functions:

```python
from sqloquent import SqlModel, HashedModel, has_one, belongs_to, contains, within

class User(SqlModel):
    ...

class Avatar(SqlModel):
    columns = ('id', 'url', 'user_id')

User.avatar = has_one('user_id', User, Avatar)
Avatar.user = belongs_to('user_id', Avatar, User)

class DAGItem(HashedModel):
    columns = ('id', 'details', 'parent_ids')

DAGItem.parents = contains('parent_ids', DAGItem, DAGItem)
DAGItem.children = within('parent_ids', DAGItem, DAGItem)
```

There are six helper functions for setting up relations between models: `has_one`,
`has_many`, `belongs_to`, `belongs_to_many`, `contains`, and `within`.
(Same are available for async, but with an `async_` prefix.)

<details>
<summary>Another example</summary>

```python
from __future__ import annotations
from sqloquent import (
    SqlModel, RelatedCollection, RelatedModel,
    has_one, has_many, belongs_to, belongs_to_many,
)

class User(SqlModel):
    table = 'users'
    columns = ('id', 'name')
    friends: RelatedCollection
    friendships: RelatedCollection
    avatar: RelatedModel
    posts: RelatedCollection

class Avatar(SqlModel):
    table = 'avatars'
    columns = ('id', 'url', 'user_id')
    user: RelatedModel

class Post(SqlModel):
    table = 'posts'
    columns = ('id', 'content', 'user_id')
    author: RelatedModel

class Friendship(SqlModel):
    table = 'friendships'
    columns = ('id', 'user1_id', 'user2_id')
    user1: RelatedModel
    user2: RelatedModel

    @classmethod
    def insert(cls, data: dict) -> Friendship | None:
        # also set inverse relationship
        result = super().insert(data)
        if result:
            super().insert({
                **data,
                'user1_id': data['user2_id'],
                'user2_id': data['user1_id'],
            })

    @classmethod
    def insert_many(cls, items: list[dict]) -> int:
        inverse = [
            {
                'user1_id': item['user2_id'],
                'user2_id': item['user1_id']
            }
            for item in items
        ]
        return super().insert_many([*items, *inverse])

    def delete(self):
        # first delete the inverse
        self.query().equal('user1_id', self.data['user2_id']).equal(
            'user2_id', self.data['user1_id']
        ).delete()
        super().delete()

User.avatar = has_one(User, Avatar)
Avatar.user = belongs_to(Avatar, User)

User.posts = has_many(User, Post)
Post.author = belongs_to(Post, User)

User.friendships = has_many(User, Friendship, 'user1_id')
User.friends = belongs_to_many(User, User, Friendship, 'user1_id', 'user2_id')

Friendship.user1 = belongs_to(Friendship, User, 'user1_id')
Friendship.user2 = belongs_to(Friendship, User, 'user2_id')
```

The relations can then be used as follows:

```python
# add users
alice: User = User.insert({"name": "Alice"})
bob: User = User.insert({"name": "Bob"})

# add avatars
alice.avatar().secondary = Avatar.insert({
    "url": "http://www.perseus.tufts.edu/img/newbanner.png",
})
alice.avatar().save()
bob.avatar = Avatar.insert({
    "url": "https://upload.wikimedia.org/wikipedia/commons/thumb/9/90"
    "/Walrus_(Odobenus_rosmarus)_on_Svalbard.jpg/1200px-Walrus_(Odobe"
    "nus_rosmarus)_on_Svalbard.jpg",
})
bob.avatar().save()

# add a friendship
bob.friends = [alice]
bob.friends().save()
bob.friendships().reload()
alice.friendships().reload()
alice.friends().reload()
```

The above is included in the second integration test:
- [models](https://github.com/k98kurz/sqloquent/blob/master/tests/integration_vectors/models2.py)
- [test](https://github.com/k98kurz/sqloquent/blob/master/tests/test_integration.py#L371)
</details>

NB: polymorphic relations are not supported. See the `Attachment` class for an
example of how to implement polymorphism if necessary.

<details>
<summary>Another example of the Contains and Within relations</summary>

```python
from sqloquent import (
    HashedModel, RelatedCollection, RelatedModel, contains, within,
)

class DAGItem(HashedModel):
    table = 'dag'
    columns = ('id', 'details', 'parent_ids')
    parents: RelatedCollection
    children: RelatedCollection

    @classmethod
    def insert(cls, data: dict) -> DAGItem|None:
        # """For better type hinting."""
        return super().insert(data)

    @classmethod
    def insert_many(cls, items: list[dict]) -> int:
        # """For better type hinting."""
        return super().insert_many(items)

DAGItem.parents = contains(DAGItem, DAGItem, 'parent_ids')
DAGItem.children = within(DAGItem, DAGItem, 'parent_ids')
```

Which can be used as follows:

```python
# create parents
parent1 = DAGItem.insert({'details': 'parent 1'})
parent2 = DAGItem.insert({'details': 'parent 2'})

# create children
child1 = DAGItem({'details': 'child 1'})
child1.parents = [parent1, parent2]
child1.parents().save()

child2 = DAGItem({'details': 'child 2'})
child2.parents = [parent1]
child2.parents().save()

# reload relation
parent1.children().reload()
parent2.children().reload()
assert len(parent1.children) == 2
assert len(parent2.children) == 1
```
</details>

#### Table Construction

The package as it stands relies upon text or varchar type `id` columns. The
`SqlModel` uses a hexadecimal uuid4 as a GUID, while the `HashedModel` uses the
sha256 of the deterministically encoded record content as a GUID. This can be
changed for use with autoincrementing int id columns by extending `SqlModel` and
overriding the `insert` and `insert_many` methods to prevent setting the id via
`cls.generate_id()`. However, this is not recommended unless the autoincrement
id can be reliably discerned from the db cursor and there are no concerns about,
say, synchronizing between instances using a CRDT.

Use one of the variants of the `sqloquent make migration` command to create a
migration scaffold, then edit the result as necessary. If you specify the
`--model name path/to/model/file` variant, the resultant source will include a
unique index on the id column and simple indices on all other columns. This will
also parse any class annotations that map to names of columns. For example,
given the following class,

```python
from sqloquent import SqlModel, Default
class Thing(SqlModel):
    table = 'things'
    columns = ('id', 'name', 'amount', 'is_nothing')
    id: str
    name: bytes|Default[b'something']
    amount: int|None
    is_nothing: bool|None|Default[True]
```

the `make migration --model` command will produce the following migration:

```python
from sqloquent import Migration, Table


def create_table_things() -> list[Table]:
    t = Table.create('things')
    t.text('id').unique()
    t.blob('name').default(b'something').index()
    t.integer('amount').nullable().index()
    t.boolean('is_nothing').default(True).nullable().index()
    ...
    return [t]

def drop_table_things() -> list[Table]:
    return [Table.drop('things')]

def migration(connection_string: str = '') -> Migration:
    migration = Migration(connection_string)
    migration.up(create_table_things)
    migration.down(drop_table_things)
    return migration
```

This should provide a decent scaffold for migrations, allowing the user of this
package to model their data first as classes if desired. If some custom SQL is
necessary, it can be added using a callback.

<details>
<summary>Example pseudocode scaffold for adding custom SQL to a migration</summary>

```python
def add_custom_sql(clauses: list[str]) -> list[str]:
    clauses.append("do something sql-y")
    return clauses

def create_table_things() -> list[Table]:
    t = Table.create('things')
    t.text('id').unique()
    t.blob('name').index()
    t.integer('amount').nullable().index()
    t.boolean('is_nothing').nullable().default(True).index()
    t.custom(add_custom_sql)
    ...
    return [t]
```
</details>

Examine the generated SQL of any migration using the
`sqloquent examine path/to/migration/file` command. The above example (less the
custom sql pseudocode) will generate the following:

```sql
/**** generated up/apply sql ****/
begin;
create table if not exists "things" ("id" text, "name" blob default (x'736f6d657468696e67'), "amount" integer, "is_nothing" boolean default True);
create unique index if not exists udx_things_id on "things" ("id");
create index if not exists idx_things_name on "things" ("name");
create index if not exists idx_things_amount on "things" ("amount");
create index if not exists idx_things_is_nothing on "things" ("is_nothing");
commit;

/**** generated down/undo sql ****/
begin;
drop table if exists "things";
commit;
```

For comprehensive documentation on the automatic migration system, including data
type compatibility, automatic migration tracking, and complete workflow examples,
see the
[Migration Guide](https://github.com/k98kurz/sqloquent/blob/v0.7.3/docs/migrations.md).

#### Model Event Hooks

As of v0.5.0, `SqlModel`, `HashedModel`, `DeletedModel`, `Attachment`,
`AsyncSqlModel`, `AsyncHashedModel`, `AsyncDeletedModel`, and `AsyncAttachment`
have an event hook system. Each has the following class methods:

- `add_hook(event: str, hook: Callable)`
- `remove_hook(event: str, hook: Callable)`
- `clear_hooks(event: str = None)`
- `invoke_hooks(event, *args, **kwargs)`

The async version of `invoke_hooks` will detect when an event handler returns a
coroutine and will await them if `parallel_events=True` is passed in (relevant
other methods as described below have `parallel_events=False` default argument).
Additionally, these methods contain checks to ensure that subclasses will have
their own `_event_hooks` dictionary to avoid conflicts with parent classes (i.e.
so that every class has its own unique event hooks).

The following events are shared by all models:

- `before_insert`
- `after_insert`
- `before_insert_many`
- `after_insert_many`
- `before_update`
- `after_update`
- `before_save`
- `after_save`
- `before_delete`
- `after_delete`
- `before_reload`
- `after_reload`

`DeletedModel` and `AsyncDeletedModel` also have the following unique events:

- `before_restore`
- `after_restore`

Callbacks will receive one positional arg (the class calling the event hook) and
the rest will be keyword args. Callbacks should accept `**kwargs` and check it
for expected values necessary for handling the event. The `kwargs` will at a
minimum contain the string event name under the key "event".

To manage these events, use the methods on the class. For example:

```python

class Thing(SqlModel):
    table = 'things'
    columns = ('id', 'name')

def make_handler(event):
    def handle_event(*args, **kwargs):
        print(f'{event} called')
    return handle_event

Thing.add_hook('before_insert', make_handler('before_insert'))
thing = Thing.insert({'name': 'Stuff'}) # prints "before_insert called" before all insert logic
Thing.insert({'name': 'Another Stuff'}, suppress_events = True) # event handler not called

Thing.add_hook('after_delete', make_handler('after_delete'))
thing.delete() # prints "after_delete called" after db operation
```

<details>
<summary>Async version</summary>

```python
class Thing(AsyncSqlModel):
    table = 'things'
    columns = ('id', 'name')

def make_handler(event):
    async def handle_event(*args, **kwargs):
        await asyncio.sleep(0.1)
        print(f'{event} called')
    return handle_event

Thing.add_hook('before_insert', make_handler('before_insert1'))
Thing.add_hook('before_insert', make_handler('before_insert2'))
Thing.add_hook('after_delete', make_handler('after_delete'))

async def test_events():
    # prints "before_insert1 called" and then "before_insert2" before all insert logic
    thing = await Thing.insert({'name': 'Stuff'})

    # prints "before_insert1 called" and "before_insert2" in unknowable order
    thing = await Thing.insert({'name': 'Stuff'}, parallel_events=True)

    # event handlers not called
    await Thing.insert({'name': 'Another Stuff'}, suppress_events = True)

    # prints "after_delete called" after db operation
    await thing.delete()
```
</details>

If you overwrite any hooked methods and use `super().hookedmethod()`, you should
add calls to `invoke_hooks` to manage events for this overwritten method
directly, and you should pass `suppress_events=True` to `super().hookedmethod()`
calls to avoid duplicate events.

<details>
<summary>Example</summary>

```python
class Thing(SqlModel):
    @classmethod
    def insert(cls, data: dict, /, *, suppress_events=False) -> Thing:
        """Overwrite for some reason, probably custom logic."""
        if not suppress_events:
            cls.invoke_hooks('before_insert', data=data)
        ...
        something = super().insert(data, suppress_events=True) # no duplicate events
        ...
        if not suppress_events:
            cls.invoke_hooks('after_insert', data=data, something=something)
        return something
```
</details>

Note that calls to `invoke_hooks` should pass all arguments other than the event
name as keyword arguments.

#### Coupling to a Different SQL Database Client

A guide for using this library with a differen SQL database client is provided
[here](https://github.com/k98kurz/sqloquent/blob/v0.7.3/docs/how_to_couple.md).

#### Using the Cryptographic Features

If a cryptographic audit trail is desirable, use an inheritance pattern to couple
the supplied classes to the desired `ModelProtocol` implementation, or simply change
the `connection_info` attribute to use with sqlite3. The simplest and recommended
way is to subclass from `HashedModel` or `AsyncHashedModel`, e.g. by executing
`sqloquent make model ModelName --hashed [--async] --columns id=str,etc` for a code
scaffold. The `DeletedModel` or `AsyncDeletedModel` will need to be monkey-patched
with the correct connection info to use `HashedModel` or subclasses.

The below example demonstrates how to monkey-patch everything together.

```python
from .dbcxm import SomeDBContextImplementation
from sqloquent import HashedModel, DeletedModel, Attachment, SqlQueryBuilder

env_db_file_path = 'some_file.db'
env_connstring = 'host=localhost,port=69,user=admin,password=admin'

# option 1: inheritance
class CustomQueryBuilder(SqlQueryBuilder):
    def __init__(self, model_or_table, **kwargs,):
        return super().__init__(model_or_table, SomeDBContextImplementation, **kwargs)

class NewModel(HashedModel, SomeDBModel):
    connection_info = env_connstring
    query_builder_class = CustomQueryBuilder

# option 2: bind the classes
HashedModel.connection_info = env_db_file_path
HashedModel.query_builder_class = CustomQueryBuilder
DeletedModel.connection_info = env_db_file_path
DeletedModel.query_builder_class = CustomQueryBuilder
Attachment.connection_info = env_db_file_path
Attachment.query_builder_class = CustomQueryBuilder
```

The latter must be done exactly once. The value supplied for `connection_info`
should be set with some environment configuration system, but here it is only
poorly mocked.

## Interfaces, Classes, Functions, and Tools

Below are lists of interfaces, classes, errors, and functions.

<details>
<summary>Interfaces</summary>

- CursorProtocol(Protocol)
- DBContextProtocol(Protocol)
- ModelProtocol(Protocol)
- JoinedModelProtocol(Protocol)
- RowProtocol(Protocol)
- QueryBuilderProtocol(Protocol)
- RelationProtocol(Protocol)
- RelatedModel(ModelProtocol)
- RelatedCollection(Protocol)
- ColumnProtocol(Protocol)
- TableProtocol(Protocol)
- MigrationProtocol(Protocol)
</details>

<details>
<summary>Classes</summary>

Classes implement the protocols or extend the classes indicated.

- SqlModel(ModelProtocol)
- SqlQueryBuilder(QueryBuilderProtocol)
- SqliteContext(DBContextProtocol)
- DeletedModel(SqlModel)
- HashedModel(SqlModel)
- Attachment(HashedModel)
- Row(RowProtocol)
- JoinedModel(JoinedModelProtocol)
- JoinSpec
- Relation(RelationProtocol)
- HasOne(Relation)
- HasMany(HasOne)
- BelongsTo(HasOne)
- BelongsToMany(Relation)
- Contains(HasMany)
- Within(HasMany)
- Column(ColumnProtocol)
- Table(TableProtocol)
- Migration(MigrationProtocol)
- AsyncSqlModel(AsyncModelProtocol)
- AsyncSqlQueryBuilder(AsyncQueryBuilderProtocol)
- AsyncSqliteContext(AsyncDBContextProtocol)
- AsyncDeletedModel(AsyncSqlModel)
- AsyncHashedModel(AsyncSqlModel)
- AsyncAttachment(AsyncHashedModel)
- AsyncJoinedModel(AsyncJoinedModelProtocol)
- AsyncRelation(AsyncRelationProtocol)
- AsyncHasOne(AsyncRelation)
- AsyncHasMany(AsyncHasOne)
- AsyncBelongsTo(AsyncHasOne)
- AsyncBelongsToMany(AsyncRelation)
- AsyncContains(AsyncHasMany)
- AsyncWithin(AsyncHasMany)
</details>

<details>
<summary>Functions</summary>

The package includes some ORM helper functions for setting up relations and some
other useful functions.

- `dynamic_sqlmodel`
- `has_one`
- `has_many`
- `belongs_to`
- `belongs_to_many`
- `contains`
- `within`
- `get_index_name`
- `async_dynamic_sqlmodel`
- `async_has_one`
- `async_has_many`
- `async_belongs_to`
- `async_belongs_to_many`
- `async_contains`
- `async_within`
</details>

<details>
<summary>Tools</summary>

The package includes a set of tools with a CLI invocation script.

- `make_migration_create`
- `make_migration_alter`
- `make_migration_drop`
- `make_migration_from_model`
- `make_migration_from_model_path`
- `publish_migrations`
- `make_model`
- `migrate`
- `rollback`
- `refresh`
- `examine`
- `automigrate`
- `autorollback`
- `autorefresh`
</details>

## More Resources

Check out the [Pycelium discord server](https://discord.gg/b2QFEJDX69). If you
experience a problem, please discuss it on the Discord server. All suggestions
for improvement are also welcome, and the best place for that is also Discord.
If you experience a bug and do not use Discord, open an issue on Github.

## Tests

Open a terminal in the root directory and run the following to set up:

```bash
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
```

For Windows, replace `source venv/bin/activate` with
`source venv/Scripts/activate` if using a POSIX-compliant shell or
`venv\Scripts\activate.bat` for command prompt.

Then run the tests with the following for Unix:

```bash
find tests -name test_*.py -print -exec python {} \;
```

Or for Windows:

```
python tests/test_async_classes.py
python tests/test_async_integration.py
python tests/test_async_relations.py
python tests/test_classes.py
python tests/test_relations.py
python tests/test_integration.py
python tests/test_migration.py
python tests/test_tools.py
```

The tests demonstrate the intended (and actual) behavior of the classes, as
well as some contrived examples of how they are used. Perusing the tests will be
informative to anyone seeking to use/break this package, especially the
integration test which demonstrates the full package. There are currently 494
unit tests + 6 e2e/integration tests.

## ISC License

Copyright (c) 2026 Jonathan Voss (k98kurz)

Permission to use, copy, modify, and/or distribute this software
for any purpose with or without fee is hereby granted, provided
that the above copyright notice and this permission notice appear in
all copies.

THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL
WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE
AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR
CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
