Getting started¶

Installation¶

Install the core library plus all optional integrations in one shot:

pip install "alchemiq[all]"

Or pick only what your project needs. The core package ships no database driver, so for PostgreSQL start from the postgres extra - it installs the asyncpg driver used by every postgresql+asyncpg:// DSN in this guide:

pip install "alchemiq[postgres]"

Extra	What it adds
`email`	`Email` field validation (`email-validator`)
`phone`	`Phone` field validation (`phonenumbers`)
`argon2`	Argon2 password hashing (`argon2-cffi`)
`bcrypt`	bcrypt password hashing (`bcrypt`) - not included in `all`
`crypto`	`Encrypted` field transparent encryption (`cryptography`)
`outbox`	Outbox/Relay worker via TaskIQ + aio-pika (`taskiq`, `taskiq-aio-pika`)
`fastapi`	FastAPI CRUD router + dependency injection (`fastapi`)
`faststream`	FastStream publisher + consumer DI (`faststream`)
`redis`	Repository-level Redis caching (`redis`)
`postgres`	Async PostgreSQL driver (`asyncpg`)
`clickhouse`	ClickHouse models + repository (`clickhouse-connect[async]`, `clickhouse-sqlalchemy`)
`migrations`	Alembic-based migration CLI (`alembic`)

Minimal end-to-end example¶

The steps below get a working database-backed service running from scratch.

Define a model. Subclass Model and declare fields using plain Python type annotations:

import alchemiq
from alchemiq import Model
from alchemiq.types import PK, Email

class User(Model):
    id: PK[int]
    name: str
    email: Email

Configure the engine. Call configure() once at startup with an async-compatible DSN:
```
alchemiq.configure("postgresql+asyncpg://user:password@localhost/mydb")
```
Create tables. Call create_all() to emit CREATE TABLE IF NOT EXISTS for every mapped model:
```
await alchemiq.create_all()
```

Use a Repository. Instantiate Repository with your model class to get a full CRUD surface:

from alchemiq import Repository

users = Repository(User)

# create a row
user = await users.create(name="Ada Lovelace", email="ada@example.com")

# fetch by primary key
found = await users.get(id=user.id)
print(found.name)  # Ada Lovelace

That is the complete setup. All queries are async/await; no synchronous mode is provided.