masq/ddd_domain
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[HOP] Initial commit (release: 0.0.0)

Browse Source
This commit is contained in:
max/sooulix 2026-08-16 10:32:42 +02:00
commit 24381de6bd
15 changed files with 1430 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
.gitignore vendored Normal file
View File

@ -0,0 +1,20 @@
*~
*.pyc
\#*
.#*
*.swp
tmp
build
dist
*.egg-info
.vscode
.noseids
.pdbrc
Backups
__pycache__
.hop/alt_config
.hop/local_config
.hop/backups/
.hop/production
.hop/.fetching
.half_orm_cli

7
.hop/config Normal file
View File

@ -0,0 +1,7 @@
[halfORM]
hop_version = 1.0.0-a32
git_origin = git@a38.benbart.fr:amsleaveamix/ddd_domain.git
devel = True
package_name = ddd_domain
with_half_orm_meta = False

9
.hop/model/metadata-0.0.0.sql Normal file
View File

@ -0,0 +1,9 @@
COPY half_orm_meta.database (id, name, description) FROM stdin;
\.
COPY half_orm_meta.hop_release (major, minor, patch, pre_release, pre_release_num, date, "time", changelog, commit, dbid, hop_release) FROM stdin;
0 0 0 2026-08-16 10:32:41+02 Initial release \N \N \N
\.
COPY half_orm_meta.hop_release_issue (num, issue_release, release_major, release_minor, release_patch, release_pre_release, release_pre_release_num, changelog) FROM stdin;
\.

255
.hop/model/schema-0.0.0.sql Normal file
View File

@ -0,0 +1,255 @@
--
-- PostgreSQL database dump
--
SET statement_timeout = 0;
SET lock_timeout = 0;
SET idle_in_transaction_session_timeout = 0;
SET client_encoding = 'UTF8';
SET standard_conforming_strings = on;
SELECT pg_catalog.set_config('search_path', '', false);
SET check_function_bodies = false;
SET xmloption = content;
SET client_min_messages = warning;
SET row_security = off;
--
-- Name: half_orm_meta; Type: SCHEMA; Schema: -; Owner: -
--
CREATE SCHEMA half_orm_meta;
--
-- Name: half_orm_meta.view; Type: SCHEMA; Schema: -; Owner: -
--
CREATE SCHEMA "half_orm_meta.view";
--
-- Name: check_database(text); Type: FUNCTION; Schema: half_orm_meta; Owner: -
--
CREATE FUNCTION half_orm_meta.check_database(old_dbid text DEFAULT NULL::text) RETURNS text
LANGUAGE plpgsql
AS $$
DECLARE
dbname text;
dbid text;
BEGIN
select current_database() into dbname;
--XXX: use a materialized view.
BEGIN
select encode(hmac(dbname, pg_read_file('hop_key'), 'sha1'), 'hex') into dbid;
EXCEPTION
when undefined_file then
raise NOTICE 'No hop_key file for the cluster. Will use % for dbid', dbname;
dbid := dbname;
END;
if old_dbid is not null and old_dbid != dbid
then
raise Exception 'Not the same database!';
end if;
return dbid;
END;
$$;
SET default_tablespace = '';
SET default_table_access_method = heap;
--
-- Name: bootstrap; Type: TABLE; Schema: half_orm_meta; Owner: -
--
CREATE TABLE half_orm_meta.bootstrap (
filename text NOT NULL,
version text NOT NULL,
executed_at timestamp without time zone DEFAULT now()
);
--
-- Name: TABLE bootstrap; Type: COMMENT; Schema: half_orm_meta; Owner: -
--
COMMENT ON TABLE half_orm_meta.bootstrap IS 'Tracks executed bootstrap scripts for data initialization';
--
-- Name: COLUMN bootstrap.filename; Type: COMMENT; Schema: half_orm_meta; Owner: -
--
COMMENT ON COLUMN half_orm_meta.bootstrap.filename IS 'Bootstrap file name (e.g., 1-init-users-0.1.0.sql)';
--
-- Name: COLUMN bootstrap.version; Type: COMMENT; Schema: half_orm_meta; Owner: -
--
COMMENT ON COLUMN half_orm_meta.bootstrap.version IS 'Release version from filename (e.g., 0.1.0)';
--
-- Name: COLUMN bootstrap.executed_at; Type: COMMENT; Schema: half_orm_meta; Owner: -
--
COMMENT ON COLUMN half_orm_meta.bootstrap.executed_at IS 'Timestamp when the script was executed';
--
-- Name: database; Type: TABLE; Schema: half_orm_meta; Owner: -
--
CREATE TABLE half_orm_meta.database (
id text NOT NULL,
name text NOT NULL,
description text
);
--
-- Name: TABLE database; Type: COMMENT; Schema: half_orm_meta; Owner: -
--
COMMENT ON TABLE half_orm_meta.database IS '
id identifies the database in the cluster. It uses the key
in hop_key.
';
--
-- Name: hop_release; Type: TABLE; Schema: half_orm_meta; Owner: -
--
CREATE TABLE half_orm_meta.hop_release (
major integer NOT NULL,
minor integer NOT NULL,
patch integer NOT NULL,
pre_release text DEFAULT ''::text NOT NULL,
pre_release_num text DEFAULT ''::text NOT NULL,
date date DEFAULT CURRENT_DATE,
"time" time(0) with time zone DEFAULT CURRENT_TIME,
changelog text,
commit text,
dbid text,
hop_release text,
CONSTRAINT hop_release_major_check CHECK ((major >= 0)),
CONSTRAINT hop_release_minor_check CHECK ((minor >= 0)),
CONSTRAINT hop_release_patch_check CHECK ((patch >= 0)),
CONSTRAINT hop_release_pre_release_check CHECK ((pre_release = ANY (ARRAY['alpha'::text, 'beta'::text, 'rc'::text, ''::text]))),
CONSTRAINT hop_release_pre_release_num_check CHECK (((pre_release_num = ''::text) OR (pre_release_num ~ '^\d+$'::text)))
);
--
-- Name: hop_release_issue; Type: TABLE; Schema: half_orm_meta; Owner: -
--
CREATE TABLE half_orm_meta.hop_release_issue (
num integer NOT NULL,
issue_release integer DEFAULT 0 NOT NULL,
release_major integer NOT NULL,
release_minor integer NOT NULL,
release_patch integer NOT NULL,
release_pre_release text NOT NULL,
release_pre_release_num text NOT NULL,
changelog text,
CONSTRAINT hop_release_issue_num_check CHECK ((num >= 0))
);
--
-- Name: hop_last_release; Type: VIEW; Schema: half_orm_meta.view; Owner: -
--
CREATE VIEW "half_orm_meta.view".hop_last_release AS
SELECT major,
minor,
patch,
pre_release,
pre_release_num,
date,
"time",
changelog,
commit
FROM half_orm_meta.hop_release
ORDER BY major DESC, minor DESC, patch DESC, pre_release DESC, pre_release_num DESC
LIMIT 1;
--
-- Name: hop_penultimate_release; Type: VIEW; Schema: half_orm_meta.view; Owner: -
--
CREATE VIEW "half_orm_meta.view".hop_penultimate_release AS
SELECT major,
minor,
patch
FROM ( SELECT hop_release.major,
hop_release.minor,
hop_release.patch
FROM half_orm_meta.hop_release
ORDER BY hop_release.major DESC, hop_release.minor DESC, hop_release.patch DESC
LIMIT 2) penultimate
ORDER BY major, minor, patch
LIMIT 1;
--
-- Name: bootstrap bootstrap_pkey; Type: CONSTRAINT; Schema: half_orm_meta; Owner: -
--
ALTER TABLE ONLY half_orm_meta.bootstrap
ADD CONSTRAINT bootstrap_pkey PRIMARY KEY (filename);
--
-- Name: database database_pkey; Type: CONSTRAINT; Schema: half_orm_meta; Owner: -
--
ALTER TABLE ONLY half_orm_meta.database
ADD CONSTRAINT database_pkey PRIMARY KEY (id);
--
-- Name: hop_release_issue hop_release_issue_pkey; Type: CONSTRAINT; Schema: half_orm_meta; Owner: -
--
ALTER TABLE ONLY half_orm_meta.hop_release_issue
ADD CONSTRAINT hop_release_issue_pkey PRIMARY KEY (num, issue_release);
--
-- Name: hop_release hop_release_pkey; Type: CONSTRAINT; Schema: half_orm_meta; Owner: -
--
ALTER TABLE ONLY half_orm_meta.hop_release
ADD CONSTRAINT hop_release_pkey PRIMARY KEY (major, minor, patch, pre_release, pre_release_num);
--
-- Name: hop_release hop_release_dbid_fkey; Type: FK CONSTRAINT; Schema: half_orm_meta; Owner: -
--
ALTER TABLE ONLY half_orm_meta.hop_release
ADD CONSTRAINT hop_release_dbid_fkey FOREIGN KEY (dbid) REFERENCES half_orm_meta.database(id) ON UPDATE CASCADE;
--
-- Name: hop_release_issue hop_release_issue_release_major_release_minor_release_patc_fkey; Type: FK CONSTRAINT; Schema: half_orm_meta; Owner: -
--
ALTER TABLE ONLY half_orm_meta.hop_release_issue
ADD CONSTRAINT hop_release_issue_release_major_release_minor_release_patc_fkey FOREIGN KEY (release_major, release_minor, release_patch, release_pre_release, release_pre_release_num) REFERENCES half_orm_meta.hop_release(major, minor, patch, pre_release, pre_release_num);
--
-- PostgreSQL database dump complete
--

1
.hop/model/schema.sql Symbolic link
View File

@ -0,0 +1 @@
schema-0.0.0.sql

41
.hop/releases/README.md Normal file
View File

@ -0,0 +1,41 @@
# Releases Directory
This directory manages release workflows through text files.
## Structure
```
releases/
├── 1.0.0-stage.txt # Development release (stage)
├── 1.0.0-rc.txt # Release candidate
└── 1.0.0-production.txt # Production release
```
## Release Files
Each file contains patch IDs, one per line:
```
001-initial-schema
002-add-authentication
003-user-profiles
```
## Workflow
1. **Development**: Patch development
- `half_orm dev patch create <patch-id>`
- `half_orm dev patch apply`
- `half_orm dev patch merge` (from ho-patch/<patch-id> branch)
- Patches added to X.Y.Z-patches.toml
2. **RC**: Release candidate
- `half_orm dev release promote rc`
- Creates X.Y.Z-rc.txt
- Deletes patch branches
3. **Production**: Final release
- `half_orm dev promote-to prod`
- Creates X.Y.Z-production.txt
- Apply to production: `half_orm dev deploy-to-prod`
See docs/half_orm_dev.md for complete documentation.

34
Patches/README.md Normal file
View File

@ -0,0 +1,34 @@
# Patches Directory
This directory contains schema patch files for database evolution.
## Structure
Each patch is stored in its own directory:
```
Patches/
├── 001-initial-schema/
│ ├── 01_create_users.sql
│ ├── 02_add_indexes.sql
│ └── 03_seed_data.py
├── 002-add-authentication/
│ └── 01_auth_tables.sql
```
## Workflow
1. Create release: `half_orm dev release create <level>`
2. Create patch branch: `half_orm dev patch create <patch-id>`
3. Add SQL/Python files to Patches/<patch-id>/
4. Apply patch: `half_orm dev patch apply`
5. Test your changes
6. Merge patch: `git checkout ho-patch/<patch-id> && half_orm dev patch merge`
## File Naming
- Use numeric prefixes for ordering: `01_`, `02_`, etc.
- SQL files: `*.sql`
- Python scripts: `*.py`
- Files executed in lexicographic order
See docs/half_orm_dev.md for complete documentation.

316
README.md Normal file
View File

@ -0,0 +1,316 @@
# ddd_domain
Database-driven application using [half-orm](https://github.com/half-orm/half-orm) with [half-orm-dev](https://github.com/half-orm/half-orm-dev) workflow.
**Database:** `ddd_domain`
**Generated with:** half-orm-dev==1.0.0-a32
---
## 🚀 Quick Start
### Prerequisites
```bash
# Install half-orm-dev (includes half-orm)
# IMPORTANT: Install the version specified in .hop/config.
# Replace <version> by the value of hop_version.
pip install half-orm-dev==<version>
```
**Version requirement:** This project requires `half-orm-dev >= 1.0.0-a32`
### Clone This Project
```bash
# Clone project with automatic database setup (use git_origin in .hop/config)
half_orm dev clone <git_origin>
```
This will:
- Clone the repository and checkout `ho-prod` branch
- Configure database connection in `/etc/half_orm/ddd_domain` or `${HALFORM_CONF_DIR}/ddd_domain`
- Restore production schema from `.hop/model/schema.sql`
- Generate Python ORM classes in `ddd_domain/` directory
---
## 📖 Development Workflow
This project uses **half-orm-dev** for database versioning and patch management.
### 1. Create a Release
```bash
# Create new minor release (e.g., 0.17.0)
half_orm dev release create minor
# Creates ho-release/0.17.0 branch
# Creates .hop/releases/0.17.0-patches.toml
```
### 2. Create a Patch
# Create patch (auto-added as candidate)
half_orm dev patch create <number>-<description>
# Example: Create user authentication patch
half_orm dev patch create 456-user-auth
# Creates ho-patch/456-user-auth branch
# Creates Patches/456-user-auth/ directory
```
### 3. Develop Your Changes
```bash
# Add SQL migration
echo "CREATE TABLE users (id SERIAL PRIMARY KEY, username TEXT);" > Patches/456-user-auth/1-create-users.sql
# Add Python migration (optional)
cat > Patches/456-user-auth/2-seed-data.py << 'EOF'
from ddd_domain.public.users import Users
def apply(model):
Users(username='admin').ho_insert()
EOF
# Apply patch to test database
half_orm dev patch apply
# Generated ORM classes available in ddd_domain/
from ddd_domain.public.users import Users
user = Users(username='john').ho_insert()
```
### 4. Test Your Changes
```bash
# Run tests (pytest must be configured)
pytest tests/
# half-orm-dev validates:
# - Patch SQL syntax
# - Patch idempotency (can be applied twice)
# - Test suite passes with full release context
```
### 5. Merge Patch into Release
```bash
# Merge patch (automatic validation + tests)
half_orm dev patch merge
# What happens:
# 1. Creates temp validation branch
# 2. Applies ALL staged patches + your patch
# 3. Runs pytest tests/
# 4. If tests pass → merges into ho-release/X.Y.Z
# 5. Changes patch status to "staged" in TOML
# 6. Deletes patch branch
```
### 6. Promote Release
```bash
# Stage → RC (first release candidate)
half_orm dev release promote rc
# Creates .hop/releases/0.17.0-rc1.txt snapshot
# Tags ho-release-0.17.0-rc1
# RC → Production (after validation)
half_orm dev release promote prod
# Merges into ho-prod branch
# Creates .hop/releases/0.17.0.txt
# Tags production-0.17.0
# Generates .hop/model/schema-0.17.0.sql
```
---
## 🔧 Common Commands
### Status & Information
```bash
half_orm dev check # Show repository state
```
### Patch Management
```bash
half_orm dev patch create # create a new patch for release
half_orm dev patch apply # Apply patch to dev database
half_orm dev patch merge # Integrate patch into release
```
### Release Management
```bash
half_orm dev release create <level> # Create release (patch/minor/major)
half_orm dev release promote rc # Stage → RC
half_orm dev release promote prod # RC → Production
```
### Production Deployment
```bash
# On production server
half_orm dev update # Check available releases
half_orm dev upgrade <version> # Deploy specific version
```
---
## 📁 Project Structure
```
ddd_domain/
├── .hop/ # half-orm-dev metadata
│ ├── config # Repository configuration
│ ├── model/ # Database schemas and data
│ │ ├── schema.sql # Current production schema (symlink)
│ │ ├── schema-X.Y.Z.sql # Versioned schemas
│ │ ├── metadata-X.Y.Z.sql # half_orm_meta data dumps
│ │ └── data-X.Y.Z.sql # Reference data from @HOP:data patches
│ └── releases/ # Release tracking files
│ ├── X.Y.Z-patches.toml # Development releases (mutable)
│ ├── X.Y.Z-rcN.txt # Release candidates (immutable)
│ └── X.Y.Z.txt # Production releases (immutable)
├── Patches/ # Patch directories
│ └── <id>-<description>/ # Individual patch files
│ ├── *.sql # SQL migrations (numbered)
│ └── *.py # Python migrations (optional)
├── ddd_domain/ # Generated ORM package
│ └── <schema>/ # schema classes
└── tests/ # Test suite (pytest)
```
---
## 🌿 Git Branch Structure
- **`ho-prod`**: Production branch (stable, source of truth)
- **`ho-release/X.Y.Z`**: Release integration branches (temporary)
- **`ho-patch/ID`**: Patch development branches (temporary)
All development happens on patch branches, merged into release branches, then promoted to production.
---
## 💾 Data Persistence (@HOP:data)
For reference data that must be loaded with every database installation (lookup tables, default roles, etc.), use the `@HOP:data` annotation:
```sql
-- @HOP:data
-- This file will be included in model/data-X.Y.Z.sql
INSERT INTO roles (name, description)
VALUES ('admin', 'Administrator')
ON CONFLICT (name) DO NOTHING;
INSERT INTO permissions (name)
VALUES ('read'), ('write'), ('delete')
ON CONFLICT DO NOTHING;
```
### How it works
1. **In patches**: SQL files starting with `-- @HOP:data` contain reference data
2. **Production promote**: All `@HOP:data` files are consolidated into `model/data-X.Y.Z.sql`
3. **Clone/Restore**: Data files are loaded automatically after schema restoration
4. **Production upgrade**: Data is inserted via normal patch application (no special handling)
### Best practices
- Use `ON CONFLICT DO NOTHING` or `ON CONFLICT DO UPDATE` for idempotency
- Keep data files small and focused (one concern per file)
- Number your SQL files to control execution order: `01_roles.sql`, `02_permissions.sql`
- Only use for **reference data**, not user-generated data
---
## ⚡ Async Usage
The generated package supports both synchronous and asynchronous access to the database.
### Sync
```python
from ddd_domain.public.users import Users
user = Users(username='john').ho_get()
users = list(Users(is_active=True).ho_select())
```
### Async
```python
import asyncio
from ddd_domain import aconnect, adisconnect
from ddd_domain.public.users import Users
async def main():
await aconnect()
try:
user = await Users(username='john').ho_aget()
users = await Users(is_active=True).ho_aselect()
finally:
await adisconnect()
asyncio.run(main())
```
### REST API (FastAPI)
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
import ddd_domain
@asynccontextmanager
async def lifespan(app: FastAPI):
await ddd_domain.aconnect()
yield
await ddd_domain.adisconnect()
app = FastAPI(lifespan=lifespan)
```
The async connection pool is established automatically during tests via the `_async_pool`
fixture in `tests/conftest.py`.
---
## 📚 Documentation
- **half-orm-dev**: https://github.com/half-orm/half-orm-dev
- **half-orm**: https://github.com/half-orm/half-orm
- **Workflow guide**: https://github.com/half-orm/half-orm-dev#readme
---
## 🆘 Getting Help
```bash
half_orm dev --help # General help
half_orm dev patch --help # Patch commands
half_orm dev release --help # Release commands
```
---
## ⚠️ Important Notes
1. **Always work on patch branches** - Never commit directly to ho-prod or ho-release
2. **Test before merging** - `patch merge` runs tests automatically
3. **Sequential releases** - Only the smallest version in preparation can be promoted
4. **Idempotent patches** - SQL must be replayable (use `IF NOT EXISTS`, etc.)
5. **Production schema** - `.hop/model/schema.sql` is the source of truth
---
Generated by half-orm-dev 1.0.0-a32

48
bootstrap/README.md Normal file
View File

@ -0,0 +1,48 @@
# Bootstrap Scripts
This directory contains data initialization scripts executed on empty databases.
## File Naming
Files are named with alphabetic prefixes for execution order:
- `01-init-roles.sql`
- `02-seed-config.py`
- `03-reference-data.sql`
Files are executed **alphabetically** (not numerically parsed).
## Execution Context
Bootstrap scripts run:
- **Development**: Each `patch apply` (allows iteration on bootstrap)
- **Production**: Initial `clone` only (one-time setup)
For production data changes, use **patches** (not bootstrap).
## Python Files
Python files can define a `run(model)` function to share the database connection:
```python
def run(model):
# model is the halfORM Model instance with active connection
MyModel = model.get_relation_class('schema.table')
MyModel(field='value').ho_insert()
```
Without `run(model)`, the file executes as a subprocess (must handle own connection).
## SQL Files
SQL files can use any SQL commands:
```sql
-- Initialize roles
INSERT INTO public.roles (name) VALUES ('admin'), ('user');
```
## Notes
- No tracking mechanism (files execute on every restore)
- Not idempotent (assumes empty database)
- Maintained manually by developers

33
ddd_domain/__init__.py Normal file
View File

@ -0,0 +1,33 @@
"""This module provides the model of the database for the package ddd_domain.
"""
from half_orm.model import Model
from half_orm_dev.utils import resolve_database_config_name
from pathlib import Path
_package_dir = Path(__file__).parent
_project_dir = _package_dir.parent # Go up to project root where .hop/ lives
_config_name = resolve_database_config_name(_project_dir)
MODEL = Model(_config_name, scope=__name__)
async def aconnect():
"""Establish the async connection pool.
Call once at application startup (tests: handled by conftest.py fixture).
Cannot be called at module level — requires a running event loop.
Example:
async def main():
await ddd_domain.aconnect()
try:
...
finally:
await ddd_domain.adisconnect()
"""
await MODEL.aconnect()
async def adisconnect():
"""Close the async connection pool. Call at application shutdown."""
await MODEL.adisconnect()

530
ddd_domain/ho_baseclasses.py Normal file
View File

@ -0,0 +1,530 @@
# DO NOT EDIT — auto-generated by half-orm-dev
from __future__ import annotations
from typing import Iterator, List, Optional, TYPE_CHECKING
import dataclasses
from half_orm.field import Field # type: ignore[import-not-found]
from ddd_domain import MODEL # type: ignore[import-not-found]
class DC_Relation:
# auto-generated by half-orm-dev — do not edit
@classmethod
async def ho_acopy(cls, data, columns=None) -> int: # type: ignore[empty-body]
"""Async variant of :meth:`ho_copy`. *Executes SQL.*
Requires an async connection opened with ``await model.aconnect()``.
*New in version 0.18.12.*
"""
...
async def ho_acount(self, *args, distinct: bool = False) -> int: # type: ignore[empty-body]
"""Async variant of ho_count. *Executes SQL.*
*New in version 0.18.0.*
"""
...
async def ho_adelete(self, *args, delete_all=False): # type: ignore[empty-body]
"""Async variant of ho_delete. *Executes SQL.*
*New in version 0.18.0.*
"""
...
async def ho_aget(self, *args: str) -> dict: # type: ignore[empty-body]
"""Async variant of ho_get. *Executes SQL.*
Issues a single ``SELECT … LIMIT 2`` query and returns the matching
row as a plain ``dict``.
Args:
*args: optional column names to select. If omitted, all columns
are returned.
Returns:
dict: the matching row.
Raises:
NotFoundError: no row matches the predicate.
MultipleRowsError: more than one row matches the predicate.
*New in version 1.0.0.*
"""
...
async def ho_ainsert(self, *args, upsert=False) -> dict: # type: ignore[empty-body]
"""Async variant of ho_insert. *Executes SQL.*
*New in version 0.18.0.*
"""
...
async def ho_ais_empty(self) -> bool: # type: ignore[empty-body]
"""Async variant of ho_is_empty. *Executes SQL.*
*New in version 0.18.0.*
"""
...
async def ho_aselect(self, *args, distinct: bool = False, order_by: str | None = None, limit: int | None = None, offset: int | None = None): # type: ignore[empty-body]
"""Async variant of ho_select. Returns a list of dicts (not a generator). *Executes SQL.*
*New in version 0.18.0.*
"""
...
def ho_assert_is_singleton(self): # type: ignore[empty-body]
"""Assert that this predicate identifies exactly one row, without querying the database.
A predicate is a *singleton* when:
* every field of a unique identifier (primary key or any
``UNIQUE NOT NULL`` constraint) is set with the ``=`` comparator, **or**
* a FK join constrains a unique identifier of this relation: the fields
on *this* side of the join form a PK or UNIQUE NOT NULL, and the
corresponding fields on the joined relation are all fixed with ``=``.
The check is purely structural — no SQL is executed.
Returns:
self — for chaining before a write operation.
Raises:
NotASingletonError: if no unique identifier is fully set.
Example:
ho_is_singleton usage:
```python
# OK — id is the primary key
Author(id=42).ho_assert_is_singleton()
# OK — email has a UNIQUE NOT NULL constraint
Author(email='alice@example.com').ho_assert_is_singleton()
# Raises — last_name is not a unique identifier
Author(last_name='Martin').ho_assert_is_singleton()
# OK — FK navigation: comment.post_id fixes post.id (PK)
Comment(post_id=42).fk_post().ho_assert_is_singleton()
# Typical usage: guard a single-row write
Author(id=42).ho_assert_is_singleton().ho_update(email='new@example.com')
# Via FK navigation: delete the post linked to a specific comment
Comment(post_id=42).fk_post().ho_assert_is_singleton().ho_delete()
```
*New in version 0.18.0.*
"""
...
async def ho_aupdate(self, *args, update_all=False, **kwargs): # type: ignore[empty-body]
"""Async variant of ho_update. *Executes SQL.*
*New in version 0.18.0.*
"""
...
def ho_cast(self, qrn): # type: ignore[empty-body]
"""Cast a relation to a related relation in the PostgreSQL inheritance hierarchy.
The target ``qrn`` must either be an ancestor or a descendant of this
relation in the PostgreSQL table-inheritance hierarchy. The check is
performed via the Python MRO, which :mod:`half_orm.relation_factory`
builds to mirror the PostgreSQL hierarchy.
Args:
qrn (str): qualified relation name of the target (e.g. ``'blog.event'``).
Returns:
Relation: a new instance of the target class carrying the same
field constraints and join state as ``self``.
Raises:
CastError: if ``qrn`` is not related to this relation by inheritance.
"""
...
@classmethod
def ho_copy(cls, data, columns=None) -> int: # type: ignore[empty-body]
"""Load rows into the table using PostgreSQL ``COPY FROM``. *Executes SQL.*
Much faster than repeated :meth:`ho_insert` calls for bulk loads.
No ``RETURNING`` is supported — the number of inserted rows is returned
instead.
*New in version 0.18.12.*
Args:
data: either a ``list[dict]`` (column names are taken from the keys
of the first dict) or a file-like object opened in text mode
(CSV with a header row, or headerless if ``columns`` is given).
columns (list[str] | None): explicit column list. Required when
*data* is a headerless file-like object; ignored when *data* is
a ``list[dict]``.
Returns:
int: number of rows inserted.
Raises:
ReadOnlyRelationError: if the relation is a view or other
non-writable kind.
ValueError: if *data* is empty or *columns* is required but missing.
Example:
From a list of dicts:
```python
n = Author.ho_copy([
{'first_name': 'Bob', 'last_name': 'Martin',
'birth_date': date(1980, 1, 1)},
{'first_name': 'Eve', 'last_name': 'Dupont',
'birth_date': date(1990, 5, 12)},
])
print(n) # 2
```
From a CSV file (with header row):
```python
with open('authors.csv') as f:
n = Author.ho_copy(f)
```
From a headerless CSV file:
```python
with open('authors_no_header.csv') as f:
n = Author.ho_copy(
f,
columns=['first_name', 'last_name', 'birth_date'],
)
```
"""
...
def ho_count(self, *args, distinct: bool = False): # type: ignore[empty-body]
"""Return the number of rows that satisfy the predicate. *Executes SQL.*
Args:
*args: column names for the inner SELECT (useful with
``distinct=True``).
distinct (bool): if ``True``, count only distinct tuples.
Default: ``False``.
Returns:
int: the cardinality of the extension.
Example:
ho_count usage:
```python
Author().ho_count() # total number of authors
Author(last_name='Martin').ho_count() # subset cardinality
```
*New in version 0.18.0:* ``distinct`` parameter.
"""
...
def ho_delete(self, *args, delete_all=False): # type: ignore[empty-body]
"""Remove every row that satisfies the predicate. *Executes SQL.*
Args:
*args: column names to return from the deleted rows. Pass
``'*'`` to return all columns.
delete_all (bool): must be ``True`` when no primary key field
is set, as a safety guard against accidental mass deletions.
Default: ``False``.
Returns:
list[dict] | None: the deleted rows if ``*args`` was provided,
otherwise ``None``.
Raises:
RuntimeError: if the predicate is not set and ``delete_all``
is ``False``.
Example:
ho_delete usage:
```python
# Delete one identified row
Author(id=99).ho_assert_is_singleton().ho_delete()
# Delete all posts for a given author
Post(author_id=1).ho_delete()
# Delete all posts
Post().ho_delete(delete_all=True)
```
"""
...
@classmethod
def ho_description(cls): # type: ignore[empty-body]
"""Returns the description (comment) of the relation
"""
...
def ho_dict(self): # type: ignore[empty-body]
"""Returns a dictionary containing only the values of the fields
that are set.
"""
...
def ho_freeze(self): # type: ignore[empty-body]
"""set _ho_isfrozen to True."""
...
def ho_get(self, *args: str) -> dict: # type: ignore[empty-body]
"""Fetch the single row matching this predicate from the database. *Executes SQL.*
Guarantees that the predicate matches exactly one row and returns it as
a plain ``dict`` mapping column names to their Python values.
Issues a single ``SELECT … LIMIT 2`` query.
Args:
*args: optional column names to select. If omitted, all columns
are returned.
Returns:
dict: the matching row.
Raises:
NotFoundError: no row matches the predicate.
MultipleRowsError: more than one row matches the predicate.
Example:
ho_get usage:
```python
row = Person(last_name='Lagaffe', first_name='Gaston').ho_get()
print(row['id'], row['last_name'])
```
*Changed in version 1.0.0* **(breaking)**: returns a ``dict`` instead
of a ``Relation`` object. Raises :exc:`NotFoundError` or
:exc:`MultipleRowsError` instead of the generic :exc:`ExpectedOneError`.
"""
...
def ho_insert(self, *args, upsert: bool | None = False) -> 'dict': # type: ignore[empty-body]
"""Insert the row described by this predicate. *Executes SQL.*
Args:
*args: column names to include in the returned dict. If omitted,
all columns are returned (equivalent to ``RETURNING *``).
upsert (bool): add ``ON CONFLICT DO UPDATE`` to the INSERT. Default: ``False``.
Returns:
dict: the inserted row.
Raises:
ReadOnlyRelationError: if the relation is a view or other
non-writable kind.
Example:
Insert an author:
```python
alice = Author(
first_name='Alice', last_name='Martin',
email='alice@example.com',
).ho_insert()
alice['id'] # 1
```
"""
...
def ho_is_empty(self): # type: ignore[empty-body]
"""Return ``True`` if the extension is empty, ``False`` otherwise. *Executes SQL.*
Returns:
bool
Example:
ho_is_empty usage:
```python
Author(last_name='Unknown').ho_is_empty() # True if no such author
```
"""
...
def ho_is_set(self): # type: ignore[empty-body]
"""Return True if one field at least is set or if self has been
constrained by at least one of its foreign keys or self is the
result of a combination of Relations (using set operators) where
at least one operand is itself constrained.
"""
...
def ho_mogrify(self): # type: ignore[empty-body]
"""Print the SQL SELECT that would be executed and return ``self``.
Activates SQL tracing for the next query on this object. The query
is printed to stderr when the next executor is called. Useful for
debugging predicate composition.
Returns:
self — for chaining.
Example:
```python
Author(last_name='Martin').ho_mogrify().ho_count()
```
displays:
```sql
select
count(*) from (select
r... .*
from
"blog"."author" as r...
where
(r... ."name" = 'Martin'::text)) as ho_count
```
"""
...
def ho_select(self, *args, distinct: bool = False, order_by: str | None = None, limit: int | None = None, offset: int | None = None, json_agg=None): # type: ignore[empty-body]
"""Enumerate the extension of this predicate. *Executes SQL.*
This method is a generator. Without arguments it is equivalent to
iterating directly on the relation object (``for row in rel:``).
Args:
*args: column names to project. If omitted, all columns are
returned.
distinct (bool): add ``DISTINCT`` to the SELECT. Default: ``False``.
order_by (str): SQL ``ORDER BY`` clause, e.g.
``'last_name, first_name desc'``. Default: ``None``.
limit (int): maximum number of rows to return. Default: ``None``.
offset (int): number of rows to skip. Default: ``None``.
json_agg (dict): aggregate already-set fkeys as JSON arrays via
a ``LEFT JOIN`` + ``json_agg`` + ``GROUP BY`` on the primary key.
Each entry maps a fkey attribute name to its spec:
- ``[field, ...]`` — list of column names; alias = fkey attr name.
- ``{'fields': [...], 'alias': 'name'}`` — explicit alias.
- ``[]`` — empty list returns all columns via ``row_to_json``.
- ``{'fields': [...], 'alias': 'name', 'distinct': True}`` —
deduplicate aggregated rows using a correlated subquery
(``SELECT DISTINCT … FROM … WHERE join_cond``) instead of a
LEFT JOIN. Avoids duplicates produced by intermediate JOIN
multiplications. Default: ``False``.
The fkey must have been set via ``.fk_attr.set(rel)`` before
calling ``ho_select``.
The type of the aggregated value depends on the FK direction:
- **reverse FK, non-unique** (one-to-many): a ``list`` of dicts,
empty (``[]``) when no related rows exist.
- **reverse FK, unique** (one-to-one via UNIQUE or PK constraint):
a single ``dict``, or ``None`` when no related row exists.
- **direct FK** (many-to-one): a single ``dict``, or ``None``
when the FK target is absent (nullable FK).
Yields:
dict: one row of the extension.
Example:
Project and sort:
```python
for row in Author(last_name='Martin').ho_select('id', 'email', order_by='id'):
print(row) # {'id': 1, 'email': 'alice@example.com'}
```
Aggregate related rows as JSON (reverse FK):
```python
alice = Author(last_name='Martin')
alice.post_rfk.set() # join all posts
for row in alice.ho_select(json_agg={'post_rfk': ['id', 'title']}):
print(row['post_rfk']) # [{'id': 1, 'title': '...'}, ...]
```
Chained FK (A ← B → C) — aggregate the leaf relation's data:
```python
# For each post, collect the persons who commented on it.
# post ← comment → person (comment is the junction)
post = Post(title='Hello')
comment = Comment()
comment.author_fk.set() # chain: comment → person
post.comment_rfk.set(comment)
for row in post.ho_select(json_agg={'comment_rfk': ['last_name']}):
print(row['comment_rfk']) # [{'last_name': '...'}, ...]
```
*New in version 0.18.0:* ``distinct``, ``order_by``, ``limit`` and ``offset`` parameters.
*New in version 0.18.6:* ``json_agg`` parameter.
*Changed in version 0.18.7* **(breaking)**: direct FK and singleton reverse FK (UNIQUE/PK) in ``json_agg`` return a ``dict`` (or ``None``) instead of a list.
"""
...
def ho_unaccent(self, *fields_names): # type: ignore[empty-body]
"""Sets unaccent for each field listed in fields_names"""
...
def ho_unfreeze(self): # type: ignore[empty-body]
"""Allow to add attributs to a relation"""
...
def ho_update(self, *args, update_all=False, **kwargs): # type: ignore[empty-body]
"""Update every row that satisfies the predicate. *Executes SQL.*
Args:
*args: column names to return from the updated rows. Pass
``'*'`` to return all columns. If omitted, nothing is returned.
update_all (bool): must be ``True`` when ``self`` has no
constraint set, to confirm the intent to update all rows.
Default: ``False``.
**kwargs: ``{column_name: new_value}`` pairs to apply.
``None`` values are silently ignored.
Returns:
list[dict] | None: the updated rows if ``*args`` was provided,
otherwise ``None``.
Raises:
RuntimeError: if no constraint is set and ``update_all`` is
``False``.
Example:
ho_update usage:
```python
# Update a single row — guarded by singleton check
Author(id=1).ho_assert_is_singleton().ho_update(email='new@example.com')
# Update an entire subset at once
Post(author_id=99).ho_update(content='[archived]')
```
"""
...
def ho_where_display(self): # type: ignore[empty-body]
"""Returns the predicate as a (possibly nested) dict, or ``None`` if unconstrained.
Every node in the tree — leaf, compound, or negation — always carries:
- ``'tables'``: ``set[str]`` — all ``schema.table`` names involved
(leaf: derived from the SQL AST; compound/neg: union of children).
- ``'constraints'``: ``list[dict]`` — all leaf constraints, each with
keys ``relation``, ``field``, ``comp``, ``value``
(leaf: own fields; compound/neg: concatenation of children).
A **leaf** node additionally has ``'joins'``, ``'where'``, ``'values'``.
A **compound** node (``|``, ``&``, ``-``) additionally has
``'operator'`` (``'or'``, ``'and'``, ``'and not'``),
``'left'``, and ``'right'``.
A **negation** node (``~``) additionally has ``'operator': 'neg'``
and ``'operand'``.
Returns:
dict | None: the predicate structure, or ``None`` if the relation
has no constraint set.
*New in version 0.18.0.*
"""
...

5
ddd_domain/ho_typeddicts.py Normal file
View File

@ -0,0 +1,5 @@
# TypedDicts for ddd_domain
from __future__ import annotations
from typing import TypedDict, Optional, List, Any

9
ddd_domain/sql_adapter.py Normal file
View File

@ -0,0 +1,9 @@
from half_orm.sql_adapter import SQL_ADAPTER
import typing
__SQL_ADAPTER = {
}
__SQL_ADAPTER.update({f'_{key}': value for key, value in __SQL_ADAPTER.items()})
SQL_ADAPTER = __SQL_ADAPTER

65
pyproject.toml Normal file
View File

@ -0,0 +1,65 @@
# Package for PostgreSQL ddd_domain database.
# You can edit the following parameters:
# - version (in ddd_domain/version.txt)
# - authors
# - license
# - keywords
# - description
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "ddd_domain"
dynamic = ["version"]
description = "Package for ddd_domain PostgreSQL database"
readme = "README.md"
keywords = []
authors = [
{name = "Your Name", email = "your.email@example.com"}
]
license = "MIT"
classifiers = [
# How mature is this project? Common values are
# 3 - Alpha
# 4 - Beta
# 5 - Production/Stable
"Development Status :: 3 - Alpha",
# Indicate who your project is intended for
"Intended Audience :: Developers",
"Topic :: Software Development :: Build Tools",
# Specify the Python versions you support here
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
]
requires-python = ">=3.9"
dependencies = [
"half_orm_dev==1.0.0-a32",
]
[project.urls]
Homepage = "https://a38.benbart.fr/amsleaveamix/ddd_domain"
[project.optional-dependencies]
dev = [
"pytest",
"pytest-asyncio",
]
[tool.pytest.ini_options]
pythonpath = ["."]
asyncio_mode = "auto"
[tool.setuptools.packages.find]
where = ["."]
exclude = ["contrib", "docs", "tests", "patches", "svg"]
[tool.setuptools.dynamic]
version = {file = "ddd_domain/version.txt"}

57
tests/conftest.py Normal file
View File

@ -0,0 +1,57 @@
"""Common test fixtures for ddd_domain package.
This file provides base fixtures for database integration tests.
Developers can add schema-specific or table-specific fixtures by creating
conftest.py files in the appropriate subdirectories.
Generated by half_orm v1.0.0-a32
"""
import pytest
import pytest_asyncio
from half_orm.relation import Relation
from ddd_domain import MODEL, aconnect, adisconnect
@pytest.fixture(scope="session")
def database_model():
"""
Provide database Model for integration tests (sync).
Scope: session (created once per test session)
Returns: Model instance for ddd_domain database
Example:
def test_query(database_model):
users = database_model.get_relation_class('public.users')
assert users is not None
"""
return MODEL
@pytest.fixture
def relation_class():
"""
Provide Relation base class for tests.
Returns: half_orm.relation.Relation class
Example:
def test_inheritance(relation_class):
from ddd_domain.public.users import Users
assert issubclass(Users, relation_class)
"""
return Relation
@pytest_asyncio.fixture(scope="session", autouse=True)
async def _async_pool():
"""Establish and tear down the async connection pool for the test session."""
await aconnect()
yield
await adisconnect()
try:
from .custom_conftest import *
except ImportError:
pass