Metadata-Version: 2.4
Name: databasic
Version: 0.2.1
Summary: Asynchronous database connection manager and query executor
Author: ansipunk
Author-email: ansipunk <ansipunk@proton.me>
License-Expression: MIT
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: psycopg[binary,pool]>=3,<4
Requires-Dist: sqlalchemy>=2,<2.2
Requires-Python: >=3.11, <3.16
Project-URL: Homepage, https://github.com/ansipunk/databasic
Project-URL: Documentation, https://github.com/ansipunk/databasic#readme
Project-URL: Repository, https://github.com/ansipunk/databasic
Project-URL: Issues, https://github.com/ansipunk/databasic/issues
Description-Content-Type: text/markdown

# Databasic

A small async database interface built on SQLAlchemy Core and psycopg.

SQLAlchemy Core handles query construction and compilation. Databasic handles
connection and transaction lifetimes and executes the resulting queries directly
through psycopg.

SQLAlchemy bind processors convert input values for psycopg, including JSON/JSONB
values and custom `TypeDecorator` bind processing. Returned values are decoded
by psycopg; SQLAlchemy result processors are not applied.

## Installation

```sh
uv add databasic

# or

pip install databasic
```

## Usage

```python
import sqlalchemy as sa

from databasic import Databasic


users = sa.Table(
    "users",
    sa.MetaData(),
    sa.Column("id", sa.Integer, primary_key=True),
    sa.Column("name", sa.Text, nullable=False),
)

db = Databasic("postgresql://user:password@localhost/database")

async with db:
    async with db.session() as session:
        rows = await session.fetch_all(
            users.select().where(users.c.name == "Alice")
        )
```

`Databasic` can also be connected and disconnected explicitly. Both forms have
the same semantics; explicit lifecycle management is useful when the lifetime is
controlled by an application or framework.

```python
db = Databasic("postgresql://user:password@localhost/database")

await db.connect()

try:
    async with db.session() as session:
        rows = await session.fetch_all(users.select())
finally:
    await db.disconnect()
```

A session is a transaction. It commits when the context exits successfully and
rolls back when it exits with an exception.

A previously closed database cannot be reused. This will throw an exception.

```python
db = Databasic("postgresql://user:password@localhost/database")

async with db:
    pass

# Will throw an exception.
async with db:
    pass
```

## API

### `Databasic`

```python
db = Databasic(conninfo, force_rollback=False)
```

#### `connect()`

Open the connection pool.

```python
await db.connect()
```

#### `disconnect()`

Close the connection pool.

```python
await db.disconnect()
```

#### `session()`

Create a transactional session.

```python
async with db.session() as session:
    ...
```

Successful exit commits the transaction. An exception rolls it back.

### `Session`

#### `execute()`

Execute a statement without returning rows.

```python
await session.execute(
    users.insert().values(name="Alice")
)
```

#### `execute_many()`

Execute the same statement multiple times with different parameters.

```python
query = users.insert().values(
    name=sa.bindparam("name"),
)

await session.execute_many(
    query,
    [
        {"name": "Alice"},
        {"name": "Bob"},
    ],
)
```

The statement must define its values using explicit bind parameters. Values are
supplied exclusively through the parameter mappings. Expanding parameters
(such as variable-length `IN` lists) and literal-execute parameters are not
supported by `execute_many()`.

#### `fetch_one()`

Execute a statement and return one row, or `None` if there is no row.

```python
user = await session.fetch_one(
    users.select().where(users.c.id == 1)
)
```

A row is simply a `dict[str, Any]`:

```python
{"id": 1, "name": "Alice"}
```

There is no result or row wrapper.

#### `fetch_all()`

Execute a statement and return all rows as a list of dictionaries.

```python
users = await session.fetch_all(
    users.select()
)
```

The result has the shape:

```python
[
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
]
```

#### `transaction()`

Create a nested transaction within the current session.

```python
async with db.session() as session:
    await session.execute(
        users.insert().values(name="Alice")
    )

    async with session.transaction():
        await session.execute(
            users.insert().values(name="Bob")
        )
```

Nested transactions use database savepoints. An exception rolls back the nested
transaction without rolling back earlier work in the session.

## Testing

`force_rollback=True` runs all sessions within an outer transaction that is
rolled back when Databasic is disconnected.

```python
async with Databasic(
    "postgresql://user:password@localhost/test_database",
    force_rollback=True,
) as db:
    ...
```

This allows integration tests to use normal sessions and transactions without
leaving persistent database state.
