Skip to content

Repositories

repositories

The repositories: stores addressed by identity, rather than documents read end to end.

A repository holds many plans and is asked for one - DBRepository has get, save, delete, patch and get_plan_by_id. It holds nothing but the session it was handed, so it is built where it is used and thrown away afterwards, and it never commits: the caller owns the transaction.

Connecting is separate, and happens once:

engine = create_engine("postgresql://user@host/db")
ensure_schema(engine, bootstrap=True)

with Session(engine) as session, session.begin():
    DBRepository(session).save_all(collection)

AsyncDBRepository is the same over an AsyncSession from create_async_engine; the migrations are synchronous, so an async application builds both engines and calls ensure_schema on the synchronous one.

The session must be bound to an engine from this package - xmas_core.repositories.engine says why.

A document is not a store: it has no identity to address, it is read end to end and written end to end. That is xmas_core.codec - codec.read(path) and codec.write(collection, path) - and nothing here.

AsyncDBRepository(session)

The database repository over an AsyncSession.

Shares the statement builders and the ORM-to-pydantic mapping with DBRepository as functions, not through a base class: an async def overriding a def is not a substitute for it. The methods differ in nothing but being awaited. It holds no engine of its own, so there is nothing here to close.

The migrations are synchronous, so an async application builds both engines and calls ensure_schema on the synchronous one at startup:

engine = create_engine(url)
ensure_schema(engine, bootstrap=True)
engine.dispose()

SessionLocal = async_sessionmaker(create_async_engine(url))

async with SessionLocal() as session:
    await AsyncDBRepository(session).save_all(collection)
    await session.commit()

Binds the repository to an async session for one unit of work.

Parameters:

Name Type Description Default
session AsyncSession

An AsyncSession bound to an engine from create_async_engine.

 required

Raises:

Type Description
DatasourceError

The session is unbound, its engine did not come from create_async_engine, or that engine binds no schema.
Source code in xmas_core/repositories/db_async.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(self, session: AsyncSession) -> None:
    """Binds the repository to an async session for one unit of work.

    Args:
        session: An `AsyncSession` bound to an engine from
            [`create_async_engine`][xmas_core.repositories.engine.create_async_engine].

    Raises:
        DatasourceError: The session is unbound, its engine did not come from
            [`create_async_engine`][xmas_core.repositories.engine.create_async_engine],
            or that engine binds no schema.
    """
    stmts.require_engine(session)
    self.session = session

delete(id) async

Delete one feature and return it as it was.

Source code in xmas_core/repositories/db_async.py 
113
114
115
116
117
118
async def delete(self, id: UUID) -> FeatureType:
    """Delete one feature and return it as it was."""
    logger.debug(f"deleting feature with id {id}")
    feature = stmts.require_feature(await self._load(id), id)
    await self.session.delete(feature)
    return _coretable.from_orm(feature)

delete_plan_by_id(id) async

Delete a plan and everything it owns.

The delete holds its FOR UPDATE batch until the caller commits, so delete one plan per transaction; see the synchronous method.

Source code in xmas_core/repositories/db_async.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def delete_plan_by_id(self, id: UUID) -> FeatureType:
    """Delete a plan and everything it owns.

    The delete holds its `FOR UPDATE` batch until the caller commits, so delete one
    plan per transaction; see the synchronous method.
    """
    logger.debug(f"deleting plan with id {id}")
    plan_feature = stmts.require_plan_feature(await self._load(id), id)
    plan_model = _coretable.from_orm(plan_feature)
    deleted = (
        await self.session.scalars(
            select(sql.coretable_delete_objects_recursive(plan_feature.id).id)
        )
    ).all()
    logger.debug(f"deleted {len(deleted)} feature(s) with plan {id}")
    return plan_model

get(id) async

Read one feature.

Source code in xmas_core/repositories/db_async.py 
85
86
87
88
89
async def get(self, id: UUID) -> FeatureType:
    """Read one feature."""
    logger.debug(f"retrieving feature with id {id}")
    feature = stmts.require_feature(await self._load(id), id)
    return _coretable.from_orm(feature)

get_plan_by_id(id) async

Read a plan and every feature it owns as a collection.

Source code in xmas_core/repositories/db_async.py 
73
74
75
76
77
78
79
80
81
82
83
async def get_plan_by_id(self, id: UUID) -> FeatureCollection:
    """Read a plan and every feature it owns as a collection."""
    logger.debug(f"retrieving plan with id {id}")
    plan_feature = stmts.require_plan_feature(await self._load(id), id)
    related = await self.session.scalars(
        sql.coretable_feature_graph(plan_feature.id)
        .where(Feature.id != plan_feature.id)
        # the related features are reads too, see `DBRepository`
        .execution_options(populate_existing=True)
    )
    return stmts.plan_collection_from_features(plan_feature, related.all())

patch(id, partial_update) async

Apply a partial update to an existing feature and return the result.

Source code in xmas_core/repositories/db_async.py 
153
154
155
156
157
158
159
160
async def patch(self, id: UUID, partial_update: dict[str, Any]) -> FeatureType:
    """Apply a partial update to an existing feature and return the result."""
    logger.debug(f"patching feature with id {id}: {partial_update}")
    db_feature = stmts.require_feature(await self._load(id), id)
    patched_feature = stmts.build_patched_feature(db_feature, partial_update)
    for stmt in stmts.update_stmts(id, patched_feature):
        await self.session.execute(stmt)
    return patched_feature

save(feature) async

Add a new feature to the session.

Source code in xmas_core/repositories/db_async.py 
91
92
93
94
async def save(self, feature: FeatureType) -> None:
    """Add a new feature to the session."""
    logger.debug(f"saving feature with id {feature.id}")
    self.session.add(_coretable.to_orm(feature))

save_all(features) async

Insert new features and their references in bulk.

Source code in xmas_core/repositories/db_async.py 
120
121
122
123
124
125
126
127
async def save_all(
    self, features: FeatureCollection | Iterable[FeatureType]
) -> None:
    """Insert new features and their references in bulk."""
    logger.debug("saving collection")
    feature_list, refs_list = stmts.serialize_bulk_features(features)
    for stmt, params in stmts.prepare_stmts_for_save(feature_list, refs_list):
        await self.session.execute(stmt, params)

update(id, feature) async

Replace an existing feature and its references.

Source code in xmas_core/repositories/db_async.py 
145
146
147
148
149
150
151
async def update(self, id: UUID, feature: FeatureType) -> FeatureType:
    """Replace an existing feature and its references."""
    logger.debug(f"updating feature with id {id}")
    stmts.require_feature(await self._load(id), id)
    for stmt in stmts.update_stmts(id, feature):
        await self.session.execute(stmt)
    return feature

update_all(features) async

Upsert features and replace their references in bulk.

Source code in xmas_core/repositories/db_async.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def update_all(
    self, features: FeatureCollection | Iterable[FeatureType]
) -> None:
    """Upsert features and replace their references in bulk."""
    logger.debug("updating collection")
    # Chunked per feature, as in the synchronous method.
    feats = (
        features.get_features()
        if isinstance(features, FeatureCollection)
        else features
    )
    for batch in batched(feats, stmts.batch_size()):
        feature_list, refs_list = stmts.serialize_bulk_features(batch)
        for stmt in stmts.prepare_stmts_for_update(feature_list, refs_list):
            await self.session.execute(stmt)

DBRepository(session)

Repository over the coretable database, addressed by identity.

Holds nothing but the session it is handed, so it is built where it is used and thrown away afterwards - one per request, per task, per CLI command. It never commits: the caller owns the transaction, and nothing written through a repository is persisted until the caller says so.

The session must be bound to an engine from create_engine. That engine carries the configured schema, and the repository's statements have no other way of finding it - see xmas_core.repositories.engine. Bringing the schema to head is ensure_schema, called once at startup rather than every time a repository is built.

engine = create_engine("postgresql://user@host/db")
ensure_schema(engine, bootstrap=True)

with Session(engine) as session, session.begin():
    DBRepository(session).save_all(collection)

Three consequences of the caller owning the session are worth knowing:

  • A failure inside a method - save's duplicate id, say - surfaces at the caller's flush or commit and poisons the whole transaction, rather than being confined to one call.
  • The methods autoflush, so pending work the caller has of its own is flushed from inside them; an IntegrityError that has nothing to do with this repository can surface out of get.
  • Every read goes to the database rather than to the session's identity map (populate_existing). The writes here are Core DML, which synchronizes nothing, so an update followed by a patch in one transaction would otherwise patch the pre-update state. It is also what a fresh session per call used to give for free.

Binds the repository to a session for the duration of one unit of work.

Parameters:

Name Type Description Default
session Session

A session bound to an engine from create_engine.

 required

Raises:

Type Description
DatasourceError

The session is unbound, its engine did not come from create_engine, or that engine binds no schema.
Source code in xmas_core/repositories/db.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def __init__(self, session: Session) -> None:
    """Binds the repository to a session for the duration of one unit of work.

    Args:
        session: A session bound to an engine from
            [`create_engine`][xmas_core.repositories.engine.create_engine].

    Raises:
        DatasourceError: The session is unbound, its engine did not come from
            [`create_engine`][xmas_core.repositories.engine.create_engine], or that
            engine binds no schema.
    """
    stmts.require_engine(session)
    self.session = session

delete(id)

Delete one feature and return it as it was.

Source code in xmas_core/repositories/db.py 
123
124
125
126
127
128
def delete(self, id: UUID) -> FeatureType:
    """Delete one feature and return it as it was."""
    logger.debug(f"deleting feature with id {id}")
    feature = stmts.require_feature(self._load(id), id)
    self.session.delete(feature)
    return _coretable.from_orm(feature)

delete_plan_by_id(id)

Delete a plan and everything it owns.

The delete holds its FOR UPDATE batch until the caller commits, so delete one plan per transaction: several in one transaction hold several independently ordered lock batches at once, which is the interleaving repositories.sql warns about.

Source code in xmas_core/repositories/db.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def delete_plan_by_id(self, id: UUID) -> FeatureType:
    """Delete a plan and everything it owns.

    The delete holds its `FOR UPDATE` batch until the caller commits, so delete one
    plan per transaction: several in one transaction hold several independently
    ordered lock batches at once, which is the interleaving `repositories.sql` warns about.
    """
    logger.debug(f"deleting plan with id {id}")
    plan_feature = stmts.require_plan_feature(self._load(id), id)
    plan_model = _coretable.from_orm(plan_feature)
    deleted = self.session.scalars(
        select(sql.coretable_delete_objects_recursive(plan_feature.id).id)
    ).all()
    logger.debug(f"deleted {len(deleted)} feature(s) with plan {id}")
    return plan_model

get(id)

Read one feature.

Source code in xmas_core/repositories/db.py 
 96
 97
 98
 99
100
def get(self, id: UUID) -> FeatureType:
    """Read one feature."""
    logger.debug(f"retrieving feature with id {id}")
    feature = stmts.require_feature(self._load(id), id)
    return _coretable.from_orm(feature)

get_plan_by_id(id)

Read a plan and every feature it owns as a collection.

Source code in xmas_core/repositories/db.py 
84
85
86
87
88
89
90
91
92
93
94
def get_plan_by_id(self, id: UUID) -> FeatureCollection:
    """Read a plan and every feature it owns as a collection."""
    logger.debug(f"retrieving plan with id {id}")
    plan_feature = stmts.require_plan_feature(self._load(id), id)
    related = self.session.scalars(
        sql.coretable_feature_graph(plan_feature.id)
        .where(Feature.id != plan_feature.id)
        # the related features are reads too, see the class
        .execution_options(populate_existing=True)
    ).all()
    return stmts.plan_collection_from_features(plan_feature, related)

patch(id, partial_update)

Apply a partial update to an existing feature and return the result.

Source code in xmas_core/repositories/db.py 
161
162
163
164
165
166
167
168
def patch(self, id: UUID, partial_update: dict[str, Any]) -> FeatureType:
    """Apply a partial update to an existing feature and return the result."""
    logger.debug(f"patching feature with id {id}: {partial_update}")
    db_feature = stmts.require_feature(self._load(id), id)
    patched_feature = stmts.build_patched_feature(db_feature, partial_update)
    for stmt in stmts.update_stmts(id, patched_feature):
        self.session.execute(stmt)
    return patched_feature

save(feature)

Add a new feature to the session.

Source code in xmas_core/repositories/db.py 
102
103
104
105
def save(self, feature: FeatureType) -> None:
    """Add a new feature to the session."""
    logger.debug(f"saving feature with id {feature.id}")
    self.session.add(_coretable.to_orm(feature))

save_all(features)

Insert new features and their references in bulk.

Source code in xmas_core/repositories/db.py 
130
131
132
133
134
135
def save_all(self, features: FeatureCollection | Iterable[FeatureType]) -> None:
    """Insert new features and their references in bulk."""
    logger.debug("saving collection")
    feature_list, refs_list = stmts.serialize_bulk_features(features)
    for stmt, params in stmts.prepare_stmts_for_save(feature_list, refs_list):
        self.session.execute(stmt, params)

update(id, feature)

Replace an existing feature and its references.

Source code in xmas_core/repositories/db.py 
153
154
155
156
157
158
159
def update(self, id: UUID, feature: FeatureType) -> FeatureType:
    """Replace an existing feature and its references."""
    logger.debug(f"updating feature with id {id}")
    stmts.require_feature(self._load(id), id)
    for stmt in stmts.update_stmts(id, feature):
        self.session.execute(stmt)
    return feature

update_all(features)

Upsert features and replace their references in bulk.

Source code in xmas_core/repositories/db.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def update_all(self, features: FeatureCollection | Iterable[FeatureType]) -> None:
    """Upsert features and replace their references in bulk."""
    logger.debug("updating collection")
    # Chunk per feature so a batch's refs stay paired with the features that own them.
    # On PostgreSQL every statement now binds one array per column, so this only bounds
    # the payload of a single message (see `batch_size`).
    feats = (
        features.get_features()
        if isinstance(features, FeatureCollection)
        else features
    )
    for batch in batched(feats, stmts.batch_size()):
        feature_list, refs_list = stmts.serialize_bulk_features(batch)
        for stmt in stmts.prepare_stmts_for_update(feature_list, refs_list):
            self.session.execute(stmt)

create_async_engine(datasource)

Build the asyncpg engine an AsyncDBRepository's session is bound to.

Mirrors create_engine; the schema migrations still need a synchronous engine, so an application creates both.

Parameters:

Name Type Description Default
datasource str | URL

A connection uri or a sqlalchemy.URL.

 required

Returns:

Type Description
AsyncEngine

A PostgreSQL async engine bound to the configured schema.

Raises:

Type Description
DatasourceError

The URL could not be parsed, or names a database other than PostgreSQL.
Source code in xmas_core/repositories/engine.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def create_async_engine(datasource: str | URL) -> AsyncEngine:
    """Build the asyncpg engine an `AsyncDBRepository`'s session is bound to.

    Mirrors [`create_engine`][xmas_core.repositories.engine.create_engine]; the schema
    migrations still need a synchronous engine, so an application creates both.

    Args:
        datasource: A connection uri or a `sqlalchemy.URL`.

    Returns:
        A PostgreSQL async engine bound to the configured schema.

    Raises:
        DatasourceError: The URL could not be parsed, or names a database other than
            PostgreSQL.
    """
    url = _postgresql_url(datasource)
    kwargs, options = _engine_kwargs()
    engine = _sa_create_async_engine(
        url.set(drivername="postgresql+asyncpg"),
        connect_args={"timeout": 5.0},
        execution_options=options,
        **kwargs,
    )
    return engine

create_engine(datasource)

Build the synchronous engine a repository's session is bound to.

The engine owns a connection pool for as long as it lives; the caller that built it disposes it. CORETABLE_SCHEMA is bound to the configured schema, which for_schema rebinds per tenant.

Parameters:

Name Type Description Default
datasource str | URL

A connection uri or a sqlalchemy.URL.

 required

Returns:

Type Description
Engine

A PostgreSQL engine bound to the configured schema.

Raises:

Type Description
DatasourceError

The URL could not be parsed, or names a database other than PostgreSQL.
Source code in xmas_core/repositories/engine.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def create_engine(datasource: str | URL) -> Engine:
    """Build the synchronous engine a repository's session is bound to.

    The engine owns a connection pool for as long as it lives; the caller that built it
    disposes it. `CORETABLE_SCHEMA` is bound to the configured schema, which
    [`for_schema`][xmas_core.repositories.engine.for_schema] rebinds per tenant.

    Args:
        datasource: A connection uri or a `sqlalchemy.URL`.

    Returns:
        A PostgreSQL engine bound to the configured schema.

    Raises:
        DatasourceError: The URL could not be parsed, or names a database other than
            PostgreSQL.
    """
    url = _postgresql_url(datasource)
    kwargs, options = _engine_kwargs()
    engine = _sa_create_engine(
        url.set(drivername="postgresql+psycopg"),
        connect_args={"connect_timeout": 5},
        execution_options=options,
        **kwargs,
    )
    return engine

drop_schema(engine, schema=None)

Delete the coretable and its related tables and functions from one schema.

A schema with no alembic revision is left alone: there is nothing this created to take back. A database created before 2.0 carries objects this revision knows nothing about and is dropped at the database level (DROP SCHEMA <schema> CASCADE) instead.

Parameters:

Name Type Description Default
engine Engine

A PostgreSQL engine, as built by create_engine.

 required
schema str | None

The schema to drop from. Defaults as in ensure_schema.

 None

Raises:

Type Description
RuntimeError

The schema is at a revision this release does not know.
Source code in xmas_core/repositories/engine.py 
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
def drop_schema(engine: Engine, schema: str | None = None) -> None:
    """Delete the coretable and its related tables and functions from one schema.

    A schema with no alembic revision is left alone: there is nothing this created to take
    back. A database created before 2.0 carries objects this revision knows nothing about and
    is dropped at the database level (`DROP SCHEMA <schema> CASCADE`) instead.

    Args:
        engine: A PostgreSQL engine, as built by
            [`create_engine`][xmas_core.repositories.engine.create_engine].
        schema: The schema to drop from. Defaults as in
            [`ensure_schema`][xmas_core.repositories.engine.ensure_schema].

    Raises:
        RuntimeError: The schema is at a revision this release does not know.
    """
    logger.debug("deleting tables")
    schema = (
        schema or _schema_map(engine).get(CORETABLE_SCHEMA) or get_settings().db_schema
    )
    # A downgrade creates the version table first, so an empty schema would be left with one
    # and a missing schema would fail; this reads without creating anything. On a connection
    # of its own: the read autobegins a transaction, and alembic leaves a transaction it did
    # not begin uncommitted, so the downgrade would be rolled back when the block closes.
    with engine.connect() as conn:
        schema = _resolve_schema(conn, schema)
        revision = MigrationContext.configure(
            conn, opts={"version_table_schema": schema}
        ).get_current_revision()
    if revision is None:
        logger.debug("no revision in schema %s, nothing to drop", schema)
        return
    alembic_cfg = _alembic_config(schema)
    _refuse_unknown_revisions(
        engine, script.ScriptDirectory.from_config(alembic_cfg), schema, [revision]
    )
    with engine.connect() as conn:
        alembic_cfg.attributes["connection"] = conn
        command.downgrade(alembic_cfg, "base")

ensure_schema(engine, schema=None, *, bootstrap=False)

Check that one schema behind engine is in the state the repository assumes.

Checks that the schema is accessible, at the current revision, and that the stored SRID matches the configured one. Only with bootstrap=True does it change anything: it creates a missing schema and runs the alembic migrations - creating the tables if the schema is empty - which makes it the only call a tenant's provisioning has to make, and the only one needing a role privileged enough to create a schema. Without it a missing, empty or outdated schema is an error: a mistyped name, or an unrelated existing schema, must not become a seeded tenant just because something tried to read from it. Call it once per schema, at startup; it is not something a repository does on its own, so a short-lived repository never pays for it. Several schemas are brought to head one after another, not from threads of one process: alembic keeps its migration context in module globals.

Parameters:

Name Type Description Default
engine Engine

A PostgreSQL engine, as built by create_engine.

 required
schema str | None

The schema to bring to head. Defaults to the one engine is bound to, so ensure_schema(for_schema(engine, tenant)) migrates that tenant rather than the process default - then to the configured one, and with none configured to the connection's current_schema(), which is what the engine renders for it.

 None
bootstrap bool

Create the schema if it does not exist, and bring it to head.

 False

Raises:

Type Description
RuntimeError

The schema is missing, empty or at an older revision and bootstrap is false, the current user may not create it, the user lacks the required privilege on the schema, the stored SRID differs from the configured one, a coretable was found with no alembic revision, or the schema is at a revision this release does not know (a 1.x or pre-2.0 database).
Source code in xmas_core/repositories/engine.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def ensure_schema(
    engine: Engine, schema: str | None = None, *, bootstrap: bool = False
) -> None:
    """Check that one schema behind `engine` is in the state the repository assumes.

    Checks that the schema is accessible, at the current revision, and that the stored SRID
    matches the configured one. Only with `bootstrap=True` does it change anything: it creates
    a missing schema and runs the alembic migrations - creating the tables if the schema is
    empty - which makes it the only call a tenant's provisioning has to make, and the only one
    needing a role privileged enough to create a schema. Without it a missing, empty or
    outdated schema is an error: a mistyped name, or an unrelated existing schema, must not
    become a seeded tenant just because something tried to read from it. Call it once per
    schema, at startup; it is not something a repository does on its own, so a short-lived
    repository never pays for it. Several schemas are brought to head one after another, not from threads
    of one process: alembic keeps its migration context in module globals.

    Args:
        engine: A PostgreSQL engine, as built by
            [`create_engine`][xmas_core.repositories.engine.create_engine].
        schema: The schema to bring to head. Defaults to the one `engine` is bound to, so
            `ensure_schema(for_schema(engine, tenant))` migrates that tenant rather than the
            process default - then to the configured one, and with none configured to the
            connection's `current_schema()`, which is what the engine renders for it.
        bootstrap: Create the schema if it does not exist, and bring it to head.

    Raises:
        RuntimeError: The schema is missing, empty or at an older revision and `bootstrap`
            is false, the current user may not create it, the user lacks the required
            privilege on the schema, the stored SRID differs from the
            configured one, a coretable was found with no alembic revision, or the schema
            is at a revision this release does not know (a 1.x or pre-2.0 database).
    """
    settings = get_settings()
    schema = schema or _schema_map(engine).get(CORETABLE_SCHEMA) or settings.db_schema

    def _check_schema_accessibility(privilege: str) -> None:
        """Raises an exception if the schema does not exist or is not accessible to the current user."""
        user, allowed = conn.execute(
            text("SELECT current_user, has_schema_privilege(:schema, :privilege)"),
            {"schema": schema, "privilege": privilege},
        ).one()
        if not allowed:
            raise RuntimeError(f"User {user} lacks {privilege} on schema '{schema}'")

    def _create_schema_if_missing(name: str) -> None:
        """Create schema `name` when nothing is there yet and `bootstrap` allows it."""
        # matched against the stored name rather than parsed as an identifier, which is
        # what `to_regnamespace` would do - it answers NULL for an existing `MixedCase`
        if conn.execute(
            text("SELECT 1 FROM pg_namespace WHERE nspname = :schema"),
            {"schema": name},
        ).scalar():
            return
        if not bootstrap:
            raise RuntimeError(
                f"Schema '{name}' does not exist; create it with "
                "`xmas db create-schema` or `ensure_schema(..., bootstrap=True)`"
            )
        # only when it is genuinely missing: `CREATE SCHEMA IF NOT EXISTS` wants CREATE on
        # the *database* and is refused for lacking it even when the schema is already
        # there, which is the usual grant for a tenant that owns its own schema and nothing
        # else. `IF NOT EXISTS` is still on for two processes bootstrapping the same tenant.
        try:
            conn.execute(CreateSchema(name, if_not_exists=True))
            conn.commit()
        except ProgrammingError as e:
            raise RuntimeError(
                f"Schema '{name}' does not exist and could not be created"
            ) from e

    def _check_db_srid() -> None:
        """Raises an exception if the DB SRID differs from the configured one."""
        # unqualified, and deliberately not in `CORETABLE_SCHEMA`: PostGIS installs
        # `geometry_columns` wherever the extension went, which is not a tenant's schema
        geometry_columns = Table(
            "geometry_columns",
            MetaData(),
            Column("f_table_name"),
            Column("srid"),
            Column("f_table_schema"),
        )
        stmt = select(geometry_columns.c.srid).where(
            geometry_columns.c.f_table_name == "coretable",
            geometry_columns.c.f_table_schema == schema,
        )
        srid = conn.execute(stmt).scalar_one()
        if srid != settings.db_srid:
            raise RuntimeError(
                f"DB SRID '{srid}' and configured SRID '{settings.db_srid}' must identical"
            )

    # test for tables and revision
    with engine.connect() as conn:
        schema = _resolve_schema(conn, schema)
        alembic_cfg = _alembic_config(schema)
        script_dir = script.ScriptDirectory.from_config(alembic_cfg)
        current_version = script_dir.get_heads()
        _create_schema_if_missing(schema)
        _check_schema_accessibility("USAGE")
        inspector = inspect(conn)
        tables = inspector.get_table_names(schema=schema)
        # a partial schema counts as existing data: `refs` alone missing used to
        # route a populated database into the creation branch below
        existing_coretable = {"coretable", "refs"} & set(tables)
        if "alembic_version" in tables:
            alembic_table = Table(
                "alembic_version",
                MetaData(schema=schema),
                Column("version_num"),
            )
            stmt = select(alembic_table.c.version_num)
            db_version = conn.execute(stmt).scalars().all()
        else:
            db_version = []
        _refuse_unknown_revisions(engine, script_dir, schema, db_version)
        if db_version:
            _check_db_srid()
        is_current_version = set(db_version) == set(current_version)
        if is_current_version:
            logger.info("Database is at current revision")
            return
        if bootstrap:
            _check_schema_accessibility("CREATE")
    # handle schema upgrade or table creation
    if existing_coretable and not db_version:
        e = RuntimeError("Coretable with no revision found in database")
        e.add_note(
            "it is likely that the database was set up with an older version of this library which didn't use revisions yet"
        )
        e.add_note(
            "please set up a new database or add a revision corresponding to the current model manually"
        )
        e.add_note(f"found table(s): {', '.join(sorted(existing_coretable))}")
        if existing_coretable != {"coretable", "refs"}:
            e.add_note(
                "the schema is incomplete and is never completed automatically; note "
                "that an ogr2ogr/QGIS export of a coretable is not a coretable "
                "database - it carries no refs table and renames the geometry column"
            )
        raise e
    if not bootstrap:
        raise RuntimeError(
            f"Schema {schema!r} is not at the current revision "
            f"({', '.join(db_version) or 'empty'}); bring it to head with "
            "`xmas db create-schema` or `ensure_schema(..., bootstrap=True)`"
        )
    # bring the schema to head, creating the tables if it is empty
    logger.info(
        "Running database migrations" if db_version else "Creating new database schema"
    )
    with engine.connect() as conn:
        alembic_cfg.attributes["connection"] = conn
        command.upgrade(alembic_cfg, "head")

for_schema(engine, schema) 

for_schema(engine: Engine, schema: str) -> Engine

for_schema(engine: AsyncEngine, schema: str) -> AsyncEngine

Bind an engine to one schema, sharing its pool.

This is how one process serves several tenants: the returned engine is a shallow view over the same pool, differing only in which schema CORETABLE_SCHEMA resolves to. The compiled statement is identical for every tenant - SQLAlchemy substitutes the schema after compilation - so the statement cache is shared and cannot leak between them.

Parameters:

Name Type Description Default
engine Engine | AsyncEngine

An engine from create_engine or create_async_engine.

 required
schema str

The schema to address. SQLAlchemy quotes it where it substitutes it.

 required

Returns:

Type Description
Engine | AsyncEngine

An engine of the same kind, bound to schema.
Source code in xmas_core/repositories/engine.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def for_schema(engine: Engine | AsyncEngine, schema: str) -> Engine | AsyncEngine:
    """Bind an engine to one schema, sharing its pool.

    This is how one process serves several tenants: the returned engine is a shallow view over
    the same pool, differing only in which schema `CORETABLE_SCHEMA` resolves to. The compiled
    statement is identical for every tenant - SQLAlchemy substitutes the schema after
    compilation - so the statement cache is shared and cannot leak between them.

    Args:
        engine: An engine from `create_engine` or `create_async_engine`.
        schema: The schema to address. SQLAlchemy quotes it where it substitutes it.

    Returns:
        An engine of the same kind, bound to `schema`.
    """
    return engine.execution_options(schema_translate_map={CORETABLE_SCHEMA: schema})

engine

Connecting to the coretable database, and bringing its schema to head.

Everything here is about the connection, not about the store: building an engine the repository can be handed a session from, and putting the schema behind it in the state the repository assumes. A DBRepository holds a caller-owned session and nothing else, so none of this belongs on it - constructing a repository used to run alembic, which meant a per-request repository would have migrated per request.

These are functions, not a class. SQLAlchemy's Session/sessionmaker already is the context manager, and an engine is disposed by the caller that built it, so there is nothing here to instantiate or to scope with with:

engine = create_engine(url)
try:
    ensure_schema(engine, bootstrap=True)
    with Session(engine) as session, session.begin():
        DBRepository(session).save_all(collection)
finally:
    engine.dispose()

create_engine is the one place a non-PostgreSQL database is refused, and the reason everything downstream may assume PostgreSQL. make_url accepts any dialect, and a sqlalchemy.URL handed in by a caller was never parsed from a string at all, so this is the only check a sqlite:///x ever meets. The legacy postgres:// that SQLAlchemy dropped in 1.4 takes the same exit rather than a NoSuchModuleError.

One engine, many schemas

The ORM tables are declared in the symbolic schema CORETABLE_SCHEMA, which is not a real schema and has to be bound to one before a statement can run. create_engine binds it to the configured schema - or, with none configured, to None, which SQLAlchemy renders as the connection's current_schema() (the first existing schema on its search_path), read once when the engine first connects - and for_schema rebinds it per tenant over the same pool:

engine = create_engine(url)                       # one pool
ensure_schema(engine, "tenant_a", bootstrap=True)  # once per tenant, at deployment
with Session(for_schema(engine, "tenant_a")) as session:
    DBRepository(session).get(feature_id)

Binding happens through schema_translate_map, which SQLAlchemy substitutes after a statement is compiled, so one compiled statement serves every tenant and the cache is neither fragmented nor shared wrongly.

Both kinds of statement the repository emits carry the token: the tables, and the calls to the plpgsql helpers, which xmas_core.repositories.sql qualifies with the same symbolic schema. Each schema holds its own copy of the helpers, reading that schema's tables, so the map is the one thing that decides where a statement lands; search_path plays no part beyond choosing the default schema at first connect, and this module never sets it. A session bound to no schema addresses xmas_schema_placeholder.coretable and xmas_schema_placeholder.coretable_role_graph_ids(...), neither of which exists - a loud failure rather than a silent write into another tenant's copy.

Because none of this is visible in a statement, every engine built here is stamped with ENGINE_SCHEMA_OPTION. DBRepository reads it back off its session's bind, so an engine built any other way - or one carrying no schema binding - is refused at construction.

create_async_engine(datasource)

Build the asyncpg engine an AsyncDBRepository's session is bound to.

Mirrors create_engine; the schema migrations still need a synchronous engine, so an application creates both.

Parameters:

Name Type Description Default
datasource str | URL

A connection uri or a sqlalchemy.URL.

 required

Returns:

Type Description
AsyncEngine

A PostgreSQL async engine bound to the configured schema.

Raises:

Type Description
DatasourceError

The URL could not be parsed, or names a database other than PostgreSQL.
Source code in xmas_core/repositories/engine.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def create_async_engine(datasource: str | URL) -> AsyncEngine:
    """Build the asyncpg engine an `AsyncDBRepository`'s session is bound to.

    Mirrors [`create_engine`][xmas_core.repositories.engine.create_engine]; the schema
    migrations still need a synchronous engine, so an application creates both.

    Args:
        datasource: A connection uri or a `sqlalchemy.URL`.

    Returns:
        A PostgreSQL async engine bound to the configured schema.

    Raises:
        DatasourceError: The URL could not be parsed, or names a database other than
            PostgreSQL.
    """
    url = _postgresql_url(datasource)
    kwargs, options = _engine_kwargs()
    engine = _sa_create_async_engine(
        url.set(drivername="postgresql+asyncpg"),
        connect_args={"timeout": 5.0},
        execution_options=options,
        **kwargs,
    )
    return engine

create_engine(datasource)

Build the synchronous engine a repository's session is bound to.

The engine owns a connection pool for as long as it lives; the caller that built it disposes it. CORETABLE_SCHEMA is bound to the configured schema, which for_schema rebinds per tenant.

Parameters:

Name Type Description Default
datasource str | URL

A connection uri or a sqlalchemy.URL.

 required

Returns:

Type Description
Engine

A PostgreSQL engine bound to the configured schema.

Raises:

Type Description
DatasourceError

The URL could not be parsed, or names a database other than PostgreSQL.
Source code in xmas_core/repositories/engine.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def create_engine(datasource: str | URL) -> Engine:
    """Build the synchronous engine a repository's session is bound to.

    The engine owns a connection pool for as long as it lives; the caller that built it
    disposes it. `CORETABLE_SCHEMA` is bound to the configured schema, which
    [`for_schema`][xmas_core.repositories.engine.for_schema] rebinds per tenant.

    Args:
        datasource: A connection uri or a `sqlalchemy.URL`.

    Returns:
        A PostgreSQL engine bound to the configured schema.

    Raises:
        DatasourceError: The URL could not be parsed, or names a database other than
            PostgreSQL.
    """
    url = _postgresql_url(datasource)
    kwargs, options = _engine_kwargs()
    engine = _sa_create_engine(
        url.set(drivername="postgresql+psycopg"),
        connect_args={"connect_timeout": 5},
        execution_options=options,
        **kwargs,
    )
    return engine

drop_schema(engine, schema=None)

Delete the coretable and its related tables and functions from one schema.

A schema with no alembic revision is left alone: there is nothing this created to take back. A database created before 2.0 carries objects this revision knows nothing about and is dropped at the database level (DROP SCHEMA <schema> CASCADE) instead.

Parameters:

Name Type Description Default
engine Engine

A PostgreSQL engine, as built by create_engine.

 required
schema str | None

The schema to drop from. Defaults as in ensure_schema.

 None

Raises:

Type Description
RuntimeError

The schema is at a revision this release does not know.
Source code in xmas_core/repositories/engine.py 
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
def drop_schema(engine: Engine, schema: str | None = None) -> None:
    """Delete the coretable and its related tables and functions from one schema.

    A schema with no alembic revision is left alone: there is nothing this created to take
    back. A database created before 2.0 carries objects this revision knows nothing about and
    is dropped at the database level (`DROP SCHEMA <schema> CASCADE`) instead.

    Args:
        engine: A PostgreSQL engine, as built by
            [`create_engine`][xmas_core.repositories.engine.create_engine].
        schema: The schema to drop from. Defaults as in
            [`ensure_schema`][xmas_core.repositories.engine.ensure_schema].

    Raises:
        RuntimeError: The schema is at a revision this release does not know.
    """
    logger.debug("deleting tables")
    schema = (
        schema or _schema_map(engine).get(CORETABLE_SCHEMA) or get_settings().db_schema
    )
    # A downgrade creates the version table first, so an empty schema would be left with one
    # and a missing schema would fail; this reads without creating anything. On a connection
    # of its own: the read autobegins a transaction, and alembic leaves a transaction it did
    # not begin uncommitted, so the downgrade would be rolled back when the block closes.
    with engine.connect() as conn:
        schema = _resolve_schema(conn, schema)
        revision = MigrationContext.configure(
            conn, opts={"version_table_schema": schema}
        ).get_current_revision()
    if revision is None:
        logger.debug("no revision in schema %s, nothing to drop", schema)
        return
    alembic_cfg = _alembic_config(schema)
    _refuse_unknown_revisions(
        engine, script.ScriptDirectory.from_config(alembic_cfg), schema, [revision]
    )
    with engine.connect() as conn:
        alembic_cfg.attributes["connection"] = conn
        command.downgrade(alembic_cfg, "base")

ensure_schema(engine, schema=None, *, bootstrap=False)

Check that one schema behind engine is in the state the repository assumes.

Checks that the schema is accessible, at the current revision, and that the stored SRID matches the configured one. Only with bootstrap=True does it change anything: it creates a missing schema and runs the alembic migrations - creating the tables if the schema is empty - which makes it the only call a tenant's provisioning has to make, and the only one needing a role privileged enough to create a schema. Without it a missing, empty or outdated schema is an error: a mistyped name, or an unrelated existing schema, must not become a seeded tenant just because something tried to read from it. Call it once per schema, at startup; it is not something a repository does on its own, so a short-lived repository never pays for it. Several schemas are brought to head one after another, not from threads of one process: alembic keeps its migration context in module globals.

Parameters:

Name Type Description Default
engine Engine

A PostgreSQL engine, as built by create_engine.

 required
schema str | None

The schema to bring to head. Defaults to the one engine is bound to, so ensure_schema(for_schema(engine, tenant)) migrates that tenant rather than the process default - then to the configured one, and with none configured to the connection's current_schema(), which is what the engine renders for it.

 None
bootstrap bool

Create the schema if it does not exist, and bring it to head.

 False

Raises:

Type Description
RuntimeError

The schema is missing, empty or at an older revision and bootstrap is false, the current user may not create it, the user lacks the required privilege on the schema, the stored SRID differs from the configured one, a coretable was found with no alembic revision, or the schema is at a revision this release does not know (a 1.x or pre-2.0 database).
Source code in xmas_core/repositories/engine.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def ensure_schema(
    engine: Engine, schema: str | None = None, *, bootstrap: bool = False
) -> None:
    """Check that one schema behind `engine` is in the state the repository assumes.

    Checks that the schema is accessible, at the current revision, and that the stored SRID
    matches the configured one. Only with `bootstrap=True` does it change anything: it creates
    a missing schema and runs the alembic migrations - creating the tables if the schema is
    empty - which makes it the only call a tenant's provisioning has to make, and the only one
    needing a role privileged enough to create a schema. Without it a missing, empty or
    outdated schema is an error: a mistyped name, or an unrelated existing schema, must not
    become a seeded tenant just because something tried to read from it. Call it once per
    schema, at startup; it is not something a repository does on its own, so a short-lived
    repository never pays for it. Several schemas are brought to head one after another, not from threads
    of one process: alembic keeps its migration context in module globals.

    Args:
        engine: A PostgreSQL engine, as built by
            [`create_engine`][xmas_core.repositories.engine.create_engine].
        schema: The schema to bring to head. Defaults to the one `engine` is bound to, so
            `ensure_schema(for_schema(engine, tenant))` migrates that tenant rather than the
            process default - then to the configured one, and with none configured to the
            connection's `current_schema()`, which is what the engine renders for it.
        bootstrap: Create the schema if it does not exist, and bring it to head.

    Raises:
        RuntimeError: The schema is missing, empty or at an older revision and `bootstrap`
            is false, the current user may not create it, the user lacks the required
            privilege on the schema, the stored SRID differs from the
            configured one, a coretable was found with no alembic revision, or the schema
            is at a revision this release does not know (a 1.x or pre-2.0 database).
    """
    settings = get_settings()
    schema = schema or _schema_map(engine).get(CORETABLE_SCHEMA) or settings.db_schema

    def _check_schema_accessibility(privilege: str) -> None:
        """Raises an exception if the schema does not exist or is not accessible to the current user."""
        user, allowed = conn.execute(
            text("SELECT current_user, has_schema_privilege(:schema, :privilege)"),
            {"schema": schema, "privilege": privilege},
        ).one()
        if not allowed:
            raise RuntimeError(f"User {user} lacks {privilege} on schema '{schema}'")

    def _create_schema_if_missing(name: str) -> None:
        """Create schema `name` when nothing is there yet and `bootstrap` allows it."""
        # matched against the stored name rather than parsed as an identifier, which is
        # what `to_regnamespace` would do - it answers NULL for an existing `MixedCase`
        if conn.execute(
            text("SELECT 1 FROM pg_namespace WHERE nspname = :schema"),
            {"schema": name},
        ).scalar():
            return
        if not bootstrap:
            raise RuntimeError(
                f"Schema '{name}' does not exist; create it with "
                "`xmas db create-schema` or `ensure_schema(..., bootstrap=True)`"
            )
        # only when it is genuinely missing: `CREATE SCHEMA IF NOT EXISTS` wants CREATE on
        # the *database* and is refused for lacking it even when the schema is already
        # there, which is the usual grant for a tenant that owns its own schema and nothing
        # else. `IF NOT EXISTS` is still on for two processes bootstrapping the same tenant.
        try:
            conn.execute(CreateSchema(name, if_not_exists=True))
            conn.commit()
        except ProgrammingError as e:
            raise RuntimeError(
                f"Schema '{name}' does not exist and could not be created"
            ) from e

    def _check_db_srid() -> None:
        """Raises an exception if the DB SRID differs from the configured one."""
        # unqualified, and deliberately not in `CORETABLE_SCHEMA`: PostGIS installs
        # `geometry_columns` wherever the extension went, which is not a tenant's schema
        geometry_columns = Table(
            "geometry_columns",
            MetaData(),
            Column("f_table_name"),
            Column("srid"),
            Column("f_table_schema"),
        )
        stmt = select(geometry_columns.c.srid).where(
            geometry_columns.c.f_table_name == "coretable",
            geometry_columns.c.f_table_schema == schema,
        )
        srid = conn.execute(stmt).scalar_one()
        if srid != settings.db_srid:
            raise RuntimeError(
                f"DB SRID '{srid}' and configured SRID '{settings.db_srid}' must identical"
            )

    # test for tables and revision
    with engine.connect() as conn:
        schema = _resolve_schema(conn, schema)
        alembic_cfg = _alembic_config(schema)
        script_dir = script.ScriptDirectory.from_config(alembic_cfg)
        current_version = script_dir.get_heads()
        _create_schema_if_missing(schema)
        _check_schema_accessibility("USAGE")
        inspector = inspect(conn)
        tables = inspector.get_table_names(schema=schema)
        # a partial schema counts as existing data: `refs` alone missing used to
        # route a populated database into the creation branch below
        existing_coretable = {"coretable", "refs"} & set(tables)
        if "alembic_version" in tables:
            alembic_table = Table(
                "alembic_version",
                MetaData(schema=schema),
                Column("version_num"),
            )
            stmt = select(alembic_table.c.version_num)
            db_version = conn.execute(stmt).scalars().all()
        else:
            db_version = []
        _refuse_unknown_revisions(engine, script_dir, schema, db_version)
        if db_version:
            _check_db_srid()
        is_current_version = set(db_version) == set(current_version)
        if is_current_version:
            logger.info("Database is at current revision")
            return
        if bootstrap:
            _check_schema_accessibility("CREATE")
    # handle schema upgrade or table creation
    if existing_coretable and not db_version:
        e = RuntimeError("Coretable with no revision found in database")
        e.add_note(
            "it is likely that the database was set up with an older version of this library which didn't use revisions yet"
        )
        e.add_note(
            "please set up a new database or add a revision corresponding to the current model manually"
        )
        e.add_note(f"found table(s): {', '.join(sorted(existing_coretable))}")
        if existing_coretable != {"coretable", "refs"}:
            e.add_note(
                "the schema is incomplete and is never completed automatically; note "
                "that an ogr2ogr/QGIS export of a coretable is not a coretable "
                "database - it carries no refs table and renames the geometry column"
            )
        raise e
    if not bootstrap:
        raise RuntimeError(
            f"Schema {schema!r} is not at the current revision "
            f"({', '.join(db_version) or 'empty'}); bring it to head with "
            "`xmas db create-schema` or `ensure_schema(..., bootstrap=True)`"
        )
    # bring the schema to head, creating the tables if it is empty
    logger.info(
        "Running database migrations" if db_version else "Creating new database schema"
    )
    with engine.connect() as conn:
        alembic_cfg.attributes["connection"] = conn
        command.upgrade(alembic_cfg, "head")

for_schema(engine, schema) 

for_schema(engine: Engine, schema: str) -> Engine

for_schema(engine: AsyncEngine, schema: str) -> AsyncEngine

Bind an engine to one schema, sharing its pool.

This is how one process serves several tenants: the returned engine is a shallow view over the same pool, differing only in which schema CORETABLE_SCHEMA resolves to. The compiled statement is identical for every tenant - SQLAlchemy substitutes the schema after compilation - so the statement cache is shared and cannot leak between them.

Parameters:

Name Type Description Default
engine Engine | AsyncEngine

An engine from create_engine or create_async_engine.

 required
schema str

The schema to address. SQLAlchemy quotes it where it substitutes it.

 required

Returns:

Type Description
Engine | AsyncEngine

An engine of the same kind, bound to schema.
Source code in xmas_core/repositories/engine.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def for_schema(engine: Engine | AsyncEngine, schema: str) -> Engine | AsyncEngine:
    """Bind an engine to one schema, sharing its pool.

    This is how one process serves several tenants: the returned engine is a shallow view over
    the same pool, differing only in which schema `CORETABLE_SCHEMA` resolves to. The compiled
    statement is identical for every tenant - SQLAlchemy substitutes the schema after
    compilation - so the statement cache is shared and cannot leak between them.

    Args:
        engine: An engine from `create_engine` or `create_async_engine`.
        schema: The schema to address. SQLAlchemy quotes it where it substitutes it.

    Returns:
        An engine of the same kind, bound to `schema`.
    """
    return engine.execution_options(schema_translate_map={CORETABLE_SCHEMA: schema})

CORETABLE_SCHEMA = 'xmas_schema_placeholder' module-attribute

Symbolic schema the coretable tables are declared in - never a real schema name.

xmas_core.repositories.engine binds it to the real one per session through schema_translate_map, which is what lets one engine address many tenant schemas over a single pool. It is a constant rather than a settings read because a settings read here is evaluated at import of this module, and a schema chosen later - as the CLI's --schema does - would arrive too late to qualify anything.

A None key would do the same job for the three tables here, but it rewrites every unqualified table in a statement, including tables the calling application owns. A named token moves these three - and the calls to the SQL functions, which xmas_core.repositories.sql qualifies with it too - and nothing else. A session with no binding fails on xmas_schema_placeholder.coretable rather than landing in another schema's copy.

DBRepository(session)

Repository over the coretable database, addressed by identity.

Holds nothing but the session it is handed, so it is built where it is used and thrown away afterwards - one per request, per task, per CLI command. It never commits: the caller owns the transaction, and nothing written through a repository is persisted until the caller says so.

The session must be bound to an engine from create_engine. That engine carries the configured schema, and the repository's statements have no other way of finding it - see xmas_core.repositories.engine. Bringing the schema to head is ensure_schema, called once at startup rather than every time a repository is built.

engine = create_engine("postgresql://user@host/db")
ensure_schema(engine, bootstrap=True)

with Session(engine) as session, session.begin():
    DBRepository(session).save_all(collection)

Three consequences of the caller owning the session are worth knowing:

  • A failure inside a method - save's duplicate id, say - surfaces at the caller's flush or commit and poisons the whole transaction, rather than being confined to one call.
  • The methods autoflush, so pending work the caller has of its own is flushed from inside them; an IntegrityError that has nothing to do with this repository can surface out of get.
  • Every read goes to the database rather than to the session's identity map (populate_existing). The writes here are Core DML, which synchronizes nothing, so an update followed by a patch in one transaction would otherwise patch the pre-update state. It is also what a fresh session per call used to give for free.

Binds the repository to a session for the duration of one unit of work.

Parameters:

Name Type Description Default
session Session

A session bound to an engine from create_engine.

 required

Raises:

Type Description
DatasourceError

The session is unbound, its engine did not come from create_engine, or that engine binds no schema.
Source code in xmas_core/repositories/db.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def __init__(self, session: Session) -> None:
    """Binds the repository to a session for the duration of one unit of work.

    Args:
        session: A session bound to an engine from
            [`create_engine`][xmas_core.repositories.engine.create_engine].

    Raises:
        DatasourceError: The session is unbound, its engine did not come from
            [`create_engine`][xmas_core.repositories.engine.create_engine], or that
            engine binds no schema.
    """
    stmts.require_engine(session)
    self.session = session

delete(id)

Delete one feature and return it as it was.

Source code in xmas_core/repositories/db.py 
123
124
125
126
127
128
def delete(self, id: UUID) -> FeatureType:
    """Delete one feature and return it as it was."""
    logger.debug(f"deleting feature with id {id}")
    feature = stmts.require_feature(self._load(id), id)
    self.session.delete(feature)
    return _coretable.from_orm(feature)

delete_plan_by_id(id)

Delete a plan and everything it owns.

The delete holds its FOR UPDATE batch until the caller commits, so delete one plan per transaction: several in one transaction hold several independently ordered lock batches at once, which is the interleaving repositories.sql warns about.

Source code in xmas_core/repositories/db.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def delete_plan_by_id(self, id: UUID) -> FeatureType:
    """Delete a plan and everything it owns.

    The delete holds its `FOR UPDATE` batch until the caller commits, so delete one
    plan per transaction: several in one transaction hold several independently
    ordered lock batches at once, which is the interleaving `repositories.sql` warns about.
    """
    logger.debug(f"deleting plan with id {id}")
    plan_feature = stmts.require_plan_feature(self._load(id), id)
    plan_model = _coretable.from_orm(plan_feature)
    deleted = self.session.scalars(
        select(sql.coretable_delete_objects_recursive(plan_feature.id).id)
    ).all()
    logger.debug(f"deleted {len(deleted)} feature(s) with plan {id}")
    return plan_model

get(id)

Read one feature.

Source code in xmas_core/repositories/db.py 
 96
 97
 98
 99
100
def get(self, id: UUID) -> FeatureType:
    """Read one feature."""
    logger.debug(f"retrieving feature with id {id}")
    feature = stmts.require_feature(self._load(id), id)
    return _coretable.from_orm(feature)

get_plan_by_id(id)

Read a plan and every feature it owns as a collection.

Source code in xmas_core/repositories/db.py 
84
85
86
87
88
89
90
91
92
93
94
def get_plan_by_id(self, id: UUID) -> FeatureCollection:
    """Read a plan and every feature it owns as a collection."""
    logger.debug(f"retrieving plan with id {id}")
    plan_feature = stmts.require_plan_feature(self._load(id), id)
    related = self.session.scalars(
        sql.coretable_feature_graph(plan_feature.id)
        .where(Feature.id != plan_feature.id)
        # the related features are reads too, see the class
        .execution_options(populate_existing=True)
    ).all()
    return stmts.plan_collection_from_features(plan_feature, related)

patch(id, partial_update)

Apply a partial update to an existing feature and return the result.

Source code in xmas_core/repositories/db.py 
161
162
163
164
165
166
167
168
def patch(self, id: UUID, partial_update: dict[str, Any]) -> FeatureType:
    """Apply a partial update to an existing feature and return the result."""
    logger.debug(f"patching feature with id {id}: {partial_update}")
    db_feature = stmts.require_feature(self._load(id), id)
    patched_feature = stmts.build_patched_feature(db_feature, partial_update)
    for stmt in stmts.update_stmts(id, patched_feature):
        self.session.execute(stmt)
    return patched_feature

save(feature)

Add a new feature to the session.

Source code in xmas_core/repositories/db.py 
102
103
104
105
def save(self, feature: FeatureType) -> None:
    """Add a new feature to the session."""
    logger.debug(f"saving feature with id {feature.id}")
    self.session.add(_coretable.to_orm(feature))

save_all(features)

Insert new features and their references in bulk.

Source code in xmas_core/repositories/db.py 
130
131
132
133
134
135
def save_all(self, features: FeatureCollection | Iterable[FeatureType]) -> None:
    """Insert new features and their references in bulk."""
    logger.debug("saving collection")
    feature_list, refs_list = stmts.serialize_bulk_features(features)
    for stmt, params in stmts.prepare_stmts_for_save(feature_list, refs_list):
        self.session.execute(stmt, params)

update(id, feature)

Replace an existing feature and its references.

Source code in xmas_core/repositories/db.py 
153
154
155
156
157
158
159
def update(self, id: UUID, feature: FeatureType) -> FeatureType:
    """Replace an existing feature and its references."""
    logger.debug(f"updating feature with id {id}")
    stmts.require_feature(self._load(id), id)
    for stmt in stmts.update_stmts(id, feature):
        self.session.execute(stmt)
    return feature

update_all(features)

Upsert features and replace their references in bulk.

Source code in xmas_core/repositories/db.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def update_all(self, features: FeatureCollection | Iterable[FeatureType]) -> None:
    """Upsert features and replace their references in bulk."""
    logger.debug("updating collection")
    # Chunk per feature so a batch's refs stay paired with the features that own them.
    # On PostgreSQL every statement now binds one array per column, so this only bounds
    # the payload of a single message (see `batch_size`).
    feats = (
        features.get_features()
        if isinstance(features, FeatureCollection)
        else features
    )
    for batch in batched(feats, stmts.batch_size()):
        feature_list, refs_list = stmts.serialize_bulk_features(batch)
        for stmt in stmts.prepare_stmts_for_update(feature_list, refs_list):
            self.session.execute(stmt)

AsyncDBRepository(session)

The database repository over an AsyncSession.

Shares the statement builders and the ORM-to-pydantic mapping with DBRepository as functions, not through a base class: an async def overriding a def is not a substitute for it. The methods differ in nothing but being awaited. It holds no engine of its own, so there is nothing here to close.

The migrations are synchronous, so an async application builds both engines and calls ensure_schema on the synchronous one at startup:

engine = create_engine(url)
ensure_schema(engine, bootstrap=True)
engine.dispose()

SessionLocal = async_sessionmaker(create_async_engine(url))

async with SessionLocal() as session:
    await AsyncDBRepository(session).save_all(collection)
    await session.commit()

Binds the repository to an async session for one unit of work.

Parameters:

Name Type Description Default
session AsyncSession

An AsyncSession bound to an engine from create_async_engine.

 required

Raises:

Type Description
DatasourceError

The session is unbound, its engine did not come from create_async_engine, or that engine binds no schema.
Source code in xmas_core/repositories/db_async.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(self, session: AsyncSession) -> None:
    """Binds the repository to an async session for one unit of work.

    Args:
        session: An `AsyncSession` bound to an engine from
            [`create_async_engine`][xmas_core.repositories.engine.create_async_engine].

    Raises:
        DatasourceError: The session is unbound, its engine did not come from
            [`create_async_engine`][xmas_core.repositories.engine.create_async_engine],
            or that engine binds no schema.
    """
    stmts.require_engine(session)
    self.session = session

delete(id) async

Delete one feature and return it as it was.

Source code in xmas_core/repositories/db_async.py 
113
114
115
116
117
118
async def delete(self, id: UUID) -> FeatureType:
    """Delete one feature and return it as it was."""
    logger.debug(f"deleting feature with id {id}")
    feature = stmts.require_feature(await self._load(id), id)
    await self.session.delete(feature)
    return _coretable.from_orm(feature)

delete_plan_by_id(id) async

Delete a plan and everything it owns.

The delete holds its FOR UPDATE batch until the caller commits, so delete one plan per transaction; see the synchronous method.

Source code in xmas_core/repositories/db_async.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def delete_plan_by_id(self, id: UUID) -> FeatureType:
    """Delete a plan and everything it owns.

    The delete holds its `FOR UPDATE` batch until the caller commits, so delete one
    plan per transaction; see the synchronous method.
    """
    logger.debug(f"deleting plan with id {id}")
    plan_feature = stmts.require_plan_feature(await self._load(id), id)
    plan_model = _coretable.from_orm(plan_feature)
    deleted = (
        await self.session.scalars(
            select(sql.coretable_delete_objects_recursive(plan_feature.id).id)
        )
    ).all()
    logger.debug(f"deleted {len(deleted)} feature(s) with plan {id}")
    return plan_model

get(id) async

Read one feature.

Source code in xmas_core/repositories/db_async.py 
85
86
87
88
89
async def get(self, id: UUID) -> FeatureType:
    """Read one feature."""
    logger.debug(f"retrieving feature with id {id}")
    feature = stmts.require_feature(await self._load(id), id)
    return _coretable.from_orm(feature)

get_plan_by_id(id) async

Read a plan and every feature it owns as a collection.

Source code in xmas_core/repositories/db_async.py 
73
74
75
76
77
78
79
80
81
82
83
async def get_plan_by_id(self, id: UUID) -> FeatureCollection:
    """Read a plan and every feature it owns as a collection."""
    logger.debug(f"retrieving plan with id {id}")
    plan_feature = stmts.require_plan_feature(await self._load(id), id)
    related = await self.session.scalars(
        sql.coretable_feature_graph(plan_feature.id)
        .where(Feature.id != plan_feature.id)
        # the related features are reads too, see `DBRepository`
        .execution_options(populate_existing=True)
    )
    return stmts.plan_collection_from_features(plan_feature, related.all())

patch(id, partial_update) async

Apply a partial update to an existing feature and return the result.

Source code in xmas_core/repositories/db_async.py 
153
154
155
156
157
158
159
160
async def patch(self, id: UUID, partial_update: dict[str, Any]) -> FeatureType:
    """Apply a partial update to an existing feature and return the result."""
    logger.debug(f"patching feature with id {id}: {partial_update}")
    db_feature = stmts.require_feature(await self._load(id), id)
    patched_feature = stmts.build_patched_feature(db_feature, partial_update)
    for stmt in stmts.update_stmts(id, patched_feature):
        await self.session.execute(stmt)
    return patched_feature

save(feature) async

Add a new feature to the session.

Source code in xmas_core/repositories/db_async.py 
91
92
93
94
async def save(self, feature: FeatureType) -> None:
    """Add a new feature to the session."""
    logger.debug(f"saving feature with id {feature.id}")
    self.session.add(_coretable.to_orm(feature))

save_all(features) async

Insert new features and their references in bulk.

Source code in xmas_core/repositories/db_async.py 
120
121
122
123
124
125
126
127
async def save_all(
    self, features: FeatureCollection | Iterable[FeatureType]
) -> None:
    """Insert new features and their references in bulk."""
    logger.debug("saving collection")
    feature_list, refs_list = stmts.serialize_bulk_features(features)
    for stmt, params in stmts.prepare_stmts_for_save(feature_list, refs_list):
        await self.session.execute(stmt, params)

update(id, feature) async

Replace an existing feature and its references.

Source code in xmas_core/repositories/db_async.py 
145
146
147
148
149
150
151
async def update(self, id: UUID, feature: FeatureType) -> FeatureType:
    """Replace an existing feature and its references."""
    logger.debug(f"updating feature with id {id}")
    stmts.require_feature(await self._load(id), id)
    for stmt in stmts.update_stmts(id, feature):
        await self.session.execute(stmt)
    return feature

update_all(features) async

Upsert features and replace their references in bulk.

Source code in xmas_core/repositories/db_async.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def update_all(
    self, features: FeatureCollection | Iterable[FeatureType]
) -> None:
    """Upsert features and replace their references in bulk."""
    logger.debug("updating collection")
    # Chunked per feature, as in the synchronous method.
    feats = (
        features.get_features()
        if isinstance(features, FeatureCollection)
        else features
    )
    for batch in batched(feats, stmts.batch_size()):
        feature_list, refs_list = stmts.serialize_bulk_features(batch)
        for stmt in stmts.prepare_stmts_for_update(feature_list, refs_list):
            await self.session.execute(stmt)

sql

SQLAlchemy expressions for the coretable PostgreSQL functions.

Every builder here returns an expression, never a statement: a set-returning function's image is a from-item and a scalar function's is a column element, and the caller writes the select() around it. That is what lets the caller choose the projection -- select(g) for whole rows, select(g.id) for ids alone -- and what lets a traversal go inside a join or an exists().

The functions are schema-qualified the way the tables are. Every call is a _CoretableFunction, whose schema renders as the symbolic orm.CORETABLE_SCHEMA; the schema_translate_map of an engine from repositories.create_engine/for_schema then substitutes and quotes the real schema at execution, exactly as it does for coretable. So one compiled statement serves every tenant, search_path plays no part, and the binding is the only thing that says which schema a statement addresses. The flip side: a builder run on an engine without that binding addresses xmas_schema_placeholder.<function> and fails with UndefinedFunction.

func.<schema>.<name> was declined. It needs the schema when the statement is built, which makes it a second source for the schema beside the binding - and each schema's functions read that schema's tables, so a disagreement deletes another tenant's rows without an error. It would also put a schema parameter on every builder, cost a compiled statement per tenant, and SQLAlchemy does not quote a package name that is mixed-case or a reserved word (TenantA.f, user.f).

coretable_feature_graph is the one exception and returns a Select, for the reason given in its docstring. coretable_delete_objects_recursive and coretable_delete_orphans_recursive delete when the caller's statement runs; see their warnings, in particular why narrowing the delete's output does not narrow the delete.

Every builder taking start accepts one root coretable id or a sequence of them, a bare id being equivalent to a one-element sequence. Ids absent from coretable, None elements and repeats are ignored; an empty sequence yields no rows.

MAX_BFS_DEPTH = 5 module-attribute

The maximum BFS traversal depth needed to reach everything belonging to a plan.

A literal rather than Appschema.max_containment_depth taken over the supported appschemas, which is what it equals: deriving it here would import every appschema module (~1.3 s) on every import of this module, and nothing else in it needs them. The test suite holds the two equal.

add_navigable_role(source_featuretype, navigable_role, target_featuretype, appschema, appschema_version, rel_direction, dependent_part=None)

Build the add_navigable_role call, which idempotently registers a navigable role.

Parameters:

Name Type Description Default
source_featuretype str

Owning/source feature type name.

 required
navigable_role str

Role (association) name linking source to target.

 required
target_featuretype str

Referenced/target feature type name.

 required
appschema str

Appschema prefix (e.g. "xplan").

 required
appschema_version str

Appschema version (e.g. "6.1").

 required
rel_direction Literal['forward', 'inverse']

Whether the refs edge runs "forward" or "inverse".

 required
dependent_part Literal['source', 'target'] | None

Ownership marker; "target" means source existentially owns target, "source" means target existentially owns source, None for a non-ownership role.

 None

Returns:

Type Description
Function[bool]

The function's BOOLEAN result as a column element; the function is scalar, so it belongs
Function[bool]

in a select list rather than a FROM. Execute with
Function[bool]

select(add_navigable_role(...)) and scalar_one().
Source code in xmas_core/repositories/sql.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def add_navigable_role(
    source_featuretype: str,
    navigable_role: str,
    target_featuretype: str,
    appschema: str,
    appschema_version: str,
    rel_direction: Literal["forward", "inverse"],
    dependent_part: Literal["source", "target"] | None = None,
) -> Function[bool]:
    """Build the ``add_navigable_role`` call, which idempotently registers a navigable role.

    Args:
        source_featuretype: Owning/source feature type name.
        navigable_role: Role (association) name linking source to target.
        target_featuretype: Referenced/target feature type name.
        appschema: Appschema prefix (e.g. ``"xplan"``).
        appschema_version: Appschema version (e.g. ``"6.1"``).
        rel_direction: Whether the refs edge runs ``"forward"`` or ``"inverse"``.
        dependent_part: Ownership marker; ``"target"`` means source existentially owns target,
            ``"source"`` means target existentially owns source, ``None`` for a non-ownership role.

    Returns:
        The function's ``BOOLEAN`` result as a column element; the function is scalar, so it belongs
        in a select list rather than a ``FROM``. Execute with
        ``select(add_navigable_role(...))`` and ``scalar_one()``.
    """
    return _CoretableFunction(
        "add_navigable_role",
        source_featuretype,
        navigable_role,
        target_featuretype,
        appschema,
        appschema_version,
        rel_direction,
        dependent_part,
        type_=Boolean(),
    )

coretable_delete_objects_recursive(start, dry_run=False)

Build the coretable_delete_objects_recursive call, which recursively deletes start.

Deletes each root together with its safe cascade closure (see coretable_role_graph_cascade). Pass every root of a batch in one call rather than looping: the closure is computed against the union of all roots, and the candidates are locked in a single ORDER BY id FOR UPDATE batch -- one ordered batch is what makes the delete deadlock-safe, and N calls in one transaction would take N separately ordered batches that can interleave.

Warning

Executing any statement built over this expression performs the deletion unless dry_run is True; run it inside a transaction. The affected rows are captured before removal, so the statement still yields them after the delete.

Projecting is safe; restricting rows is not. select(g.id) deletes exactly what select(g) deletes, and is the cheaper way to learn what was removed. But the function is plpgsql and materializes its whole result set before any WHERE/LIMIT/OFFSET of yours is applied, so a narrowed statement destroys the entire closure while reporting a subset of it. Never paginate a preview -- use dry_run=True.

Parameters:

Name Type Description Default
start UUID | Sequence[UUID]

One root coretable id to delete or a sequence of them; see the module docstring.

 required
dry_run bool

When True, return the closure without deleting anything.

 False

Returns:

Type Description
type[Feature]

Feature aliased onto the function scan, yielding the deleted (or, for dry_run,
type[Feature]

would-be-deleted) rows. Use select(g) for the rows or select(g.id) for their ids.
Source code in xmas_core/repositories/sql.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
def coretable_delete_objects_recursive(
    start: UUID | Sequence[UUID],
    dry_run: bool = False,
) -> type[Feature]:
    """Build the ``coretable_delete_objects_recursive`` call, which recursively deletes ``start``.

    Deletes each root together with its safe cascade closure (see `coretable_role_graph_cascade`).
    Pass every root of a batch in one call rather than looping: the closure is computed against the
    union of all roots, and the candidates are locked in a single ``ORDER BY id FOR UPDATE`` batch --
    one ordered batch is what makes the delete deadlock-safe, and N calls in one transaction would
    take N separately ordered batches that can interleave.

    Warning:
        Executing any statement built over this expression performs the deletion unless ``dry_run``
        is ``True``; run it inside a transaction. The affected rows are captured before removal, so
        the statement still yields them after the delete.

        **Projecting is safe; restricting rows is not.** ``select(g.id)`` deletes exactly what
        ``select(g)`` deletes, and is the cheaper way to learn what was removed. But the function is
        ``plpgsql`` and materializes its whole result set before any ``WHERE``/``LIMIT``/``OFFSET``
        of yours is applied, so a narrowed statement destroys the entire closure while reporting a
        subset of it. Never paginate a preview -- use ``dry_run=True``.

    Args:
        start: One root coretable id to delete or a sequence of them; see the module docstring.
        dry_run: When ``True``, return the closure without deleting anything.

    Returns:
        ``Feature`` aliased onto the function scan, yielding the deleted (or, for ``dry_run``,
        would-be-deleted) rows. Use ``select(g)`` for the rows or ``select(g.id)`` for their ids.
    """
    return aliased(
        Feature,
        _CoretableFunction(
            "coretable_delete_objects_recursive", _root_ids(start), dry_run
        ).table_valued(*Feature.__table__.c),
    )

coretable_delete_orphans_recursive()

Build the coretable_delete_orphans_recursive call.

Deletes dependent-part objects no longer owned by any whole.

Warning

Executing the caller's statement performs the deletion; run it inside a transaction.

Returns:

Type Description
Function[int]

The total number of deleted rows as an INTEGER column element; the function is scalar,
Function[int]

so it belongs in a select list rather than a FROM. Execute with
Function[int]

select(coretable_delete_orphans_recursive()) and scalar_one().
Source code in xmas_core/repositories/sql.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def coretable_delete_orphans_recursive() -> Function[int]:
    """Build the ``coretable_delete_orphans_recursive`` call.

    Deletes dependent-part objects no longer owned by any whole.

    Warning:
        Executing the caller's statement performs the deletion; run it inside a transaction.

    Returns:
        The total number of deleted rows as an ``INTEGER`` column element; the function is scalar,
        so it belongs in a select list rather than a ``FROM``. Execute with
        ``select(coretable_delete_orphans_recursive())`` and ``scalar_one()``.
    """
    return _CoretableFunction("coretable_delete_orphans_recursive", type_=Integer())

coretable_feature_graph(start_id, depth_limit=MAX_BFS_DEPTH, include_forward=True, include_backward=True)

Return a chainable Select of Features reachable from start_id via the role graph.

The role-graph traversal under the plan-graph defaults: one root and a bounded depth, with the ownership closure off.

This is the module's one exception to returning an expression, and the one traversal written over coretable rather than over a function scan: get_plan_by_id appends .where(Feature.id != plan_id) to the result, so it has to be chainable with a plain Feature.… predicate -- which is what selecting from real coretable here buys. Use coretable_role_graph instead where that constraint does not apply.

Parameters:

Name Type Description Default
start_id UUID

Root coretable id to traverse from.

 required
depth_limit int

Maximum BFS depth; defaults to the current appschemas' maximum depth.

 MAX_BFS_DEPTH
include_forward bool

Traverse navigable roles in the source->target direction.

 True
include_backward bool

Traverse navigable roles in the target->source direction.

 True

Returns:

Type Description
Select[tuple[Feature]]

A chainable Select yielding ORM Feature rows.
Source code in xmas_core/repositories/sql.py 
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
def coretable_feature_graph(
    start_id: UUID,
    depth_limit: int = MAX_BFS_DEPTH,
    include_forward: bool = True,
    include_backward: bool = True,
) -> Select[tuple[Feature]]:
    """Return a chainable Select of Features reachable from ``start_id`` via the role graph.

    The role-graph traversal under the plan-graph defaults: one root and a bounded depth, with the
    ownership closure off.

    This is the module's one exception to returning an expression, and the one traversal written over
    ``coretable`` rather than over a function scan: ``get_plan_by_id`` appends
    ``.where(Feature.id != plan_id)`` to the result, so it has to be chainable with a plain
    ``Feature.…`` predicate -- which is what selecting from real ``coretable`` here buys. Use
    `coretable_role_graph` instead where that constraint does not apply.

    Args:
        start_id: Root coretable id to traverse from.
        depth_limit: Maximum BFS depth; defaults to the current appschemas' maximum depth.
        include_forward: Traverse navigable roles in the source->target direction.
        include_backward: Traverse navigable roles in the target->source direction.

    Returns:
        A chainable ``Select`` yielding ORM ``Feature`` rows.
    """
    graph_ids = coretable_role_graph_ids(
        start_id, depth_limit, include_forward, include_backward, False
    )
    return select(Feature).where(exists().where(Feature.id == graph_ids.c.id))

coretable_role_graph(start, depth_limit=0, include_forward=True, include_backward=True, dependent_parts=False)

Build the coretable_role_graph call, the Feature rows reachable from start.

The SQL function is a thin SETOF wrapper joining coretable against coretable_role_graph_ids, so this hydrates exactly the traversal whose ids coretable_role_graph_ids returns.

Parameters:

Name Type Description Default
start UUID | Sequence[UUID]

One root coretable id or a sequence of them; see the module docstring.

 required
depth_limit int

Maximum BFS depth; 0 means unbounded.

 0
include_forward bool

Traverse navigable roles in the source->target direction.

 True
include_backward bool

Traverse navigable roles in the target->source direction.

 True
dependent_parts bool

Collect the transitive existential dependent-part (ownership) closure instead of a general traversal; both directions are followed, ignoring the include_* flags.

 False

Returns:

Type Description
type[Feature]

Feature aliased onto the function scan. Select whole rows with select(g), ids alone
type[Feature]

with select(g.id), and filter with select(g).where(g.… ) -- predicates must go
type[Feature]

through the returned alias, since Feature.… would name coretable and add it to the
type[Feature]

FROM clause a second time as a cross join.
type[Feature]

The annotation is type[Feature] rather than AliasedClass[Feature] because that is
type[Feature]

how SQLAlchemy declares aliased() over a mapped class (its private AliasedType
type[Feature]

alias), which is what makes attribute access on the result check against Feature. The
type[Feature]

runtime object is an AliasedClass.
Source code in xmas_core/repositories/sql.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def coretable_role_graph(
    start: UUID | Sequence[UUID],
    depth_limit: int = 0,
    include_forward: bool = True,
    include_backward: bool = True,
    dependent_parts: bool = False,
) -> type[Feature]:
    """Build the ``coretable_role_graph`` call, the ``Feature`` rows reachable from ``start``.

    The SQL function is a thin SETOF wrapper joining ``coretable`` against
    ``coretable_role_graph_ids``, so this hydrates exactly the traversal whose ids
    `coretable_role_graph_ids` returns.

    Args:
        start: One root coretable id or a sequence of them; see the module docstring.
        depth_limit: Maximum BFS depth; ``0`` means unbounded.
        include_forward: Traverse navigable roles in the source->target direction.
        include_backward: Traverse navigable roles in the target->source direction.
        dependent_parts: Collect the transitive existential dependent-part (ownership) closure
            instead of a general traversal; both directions are followed, ignoring the include_*
            flags.

    Returns:
        ``Feature`` aliased onto the function scan. Select whole rows with ``select(g)``, ids alone
        with ``select(g.id)``, and filter with ``select(g).where(g.… )`` -- predicates must go
        through the returned alias, since ``Feature.…`` would name ``coretable`` and add it to the
        ``FROM`` clause a second time as a cross join.

        The annotation is ``type[Feature]`` rather than ``AliasedClass[Feature]`` because that is
        how SQLAlchemy declares ``aliased()`` over a mapped class (its private ``AliasedType``
        alias), which is what makes attribute access on the result check against ``Feature``. The
        runtime object is an ``AliasedClass``.
    """
    return aliased(
        Feature,
        _CoretableFunction(
            "coretable_role_graph",
            _root_ids(start),
            depth_limit,
            include_forward,
            include_backward,
            dependent_parts,
        ).table_valued(*Feature.__table__.c),
    )

coretable_role_graph_cascade(start, depth_limit=0)

Build the coretable_role_graph_cascade call, the rows safe to delete with start.

The result is the dependent-part closure of start minus any part whose owning whole lies outside that closure (transitively), i.e. the objects that can be removed together with start without orphaning a part still owned from elsewhere. Every root is exempt from that exclusion, so a root that is itself an owned part is still returned.

Warning

The closure is not compositional, so a batch must be passed in one call. For a part owned by both a and b, neither cascade(a) nor cascade(b) contains it -- it is external to each closure taken alone -- and deleting both roots one at a time would leave it ownerless. cascade([a, b]) computes the closure against the union, where the part is external to neither.

Parameters:

Name Type Description Default
start UUID | Sequence[UUID]

One root coretable id or a sequence of them; see the module docstring.

 required
depth_limit int

Maximum BFS depth; 0 means unbounded.

 0

Returns:

Type Description
type[Feature]

Feature aliased onto the function scan; see coretable_role_graph on selecting and
type[Feature]

filtering through the alias.
Source code in xmas_core/repositories/sql.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def coretable_role_graph_cascade(
    start: UUID | Sequence[UUID],
    depth_limit: int = 0,
) -> type[Feature]:
    """Build the ``coretable_role_graph_cascade`` call, the rows safe to delete with ``start``.

    The result is the dependent-part closure of ``start`` minus any part whose owning whole lies
    outside that closure (transitively), i.e. the objects that can be removed together with
    ``start`` without orphaning a part still owned from elsewhere. Every root is exempt from that
    exclusion, so a root that is itself an owned part is still returned.

    Warning:
        The closure is not compositional, so a batch must be passed in one call. For a part owned
        by both ``a`` and ``b``, neither ``cascade(a)`` nor ``cascade(b)`` contains it -- it is
        external to each closure taken alone -- and deleting both roots one at a time would leave
        it ownerless. ``cascade([a, b])`` computes the closure against the union, where the part is
        external to neither.

    Args:
        start: One root coretable id or a sequence of them; see the module docstring.
        depth_limit: Maximum BFS depth; ``0`` means unbounded.

    Returns:
        ``Feature`` aliased onto the function scan; see `coretable_role_graph` on selecting and
        filtering through the alias.
    """
    return aliased(
        Feature,
        _CoretableFunction(
            "coretable_role_graph_cascade", _root_ids(start), depth_limit
        ).table_valued(*Feature.__table__.c),
    )

coretable_role_graph_ids(start, depth_limit=0, include_forward=True, include_backward=True, dependent_parts=False)

Build the coretable_role_graph_ids call, the ids reachable from start.

The cheap half of coretable_role_graph: it yields ids without joining coretable, so prefer it wherever the rows are not needed. The edge set the traversal walks is bounded by |refs| and independent of the roots, so one call over N roots builds it once where N calls build it N times.

Parameters:

Name Type Description Default
start UUID | Sequence[UUID]

One root coretable id or a sequence of them; see the module docstring.

 required
depth_limit int

Maximum BFS depth; 0 means unbounded.

 0
include_forward bool

Traverse navigable roles in the source->target direction.

 True
include_backward bool

Traverse navigable roles in the target->source direction.

 True
dependent_parts bool

Collect the transitive existential dependent-part (ownership) closure instead of a general traversal; both directions are followed, ignoring the include_* flags.

 False

Returns:

Type Description
TableValuedAlias

A derived table with a single id column: select(coretable_role_graph_ids(x).c.id)
TableValuedAlias

yields the ids, and the same expression goes inside an exists() or a join.
Source code in xmas_core/repositories/sql.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
def coretable_role_graph_ids(
    start: UUID | Sequence[UUID],
    depth_limit: int = 0,
    include_forward: bool = True,
    include_backward: bool = True,
    dependent_parts: bool = False,
) -> TableValuedAlias:
    """Build the ``coretable_role_graph_ids`` call, the ids reachable from ``start``.

    The cheap half of `coretable_role_graph`: it yields ids without joining ``coretable``, so prefer
    it wherever the rows are not needed. The edge set the traversal walks is bounded by ``|refs|``
    and independent of the roots, so one call over N roots builds it once where N calls build it N
    times.

    Args:
        start: One root coretable id or a sequence of them; see the module docstring.
        depth_limit: Maximum BFS depth; ``0`` means unbounded.
        include_forward: Traverse navigable roles in the source->target direction.
        include_backward: Traverse navigable roles in the target->source direction.
        dependent_parts: Collect the transitive existential dependent-part (ownership) closure
            instead of a general traversal; both directions are followed, ignoring the include_*
            flags.

    Returns:
        A derived table with a single ``id`` column: ``select(coretable_role_graph_ids(x).c.id)``
        yields the ids, and the same expression goes inside an ``exists()`` or a join.
    """
    return _CoretableFunction(
        "coretable_role_graph_ids",
        _root_ids(start),
        depth_limit,
        include_forward,
        include_backward,
        dependent_parts,
    ).table_valued("id")

list_navigable_roles()

Build the list_navigable_roles call, which returns every configured navigable role.

Returns:

Type Description
TableValuedAlias

A derived table with one row per configured role and columns source_featuretype,
TableValuedAlias

navigable_role, target_featuretype, appschema, appschema_version,
TableValuedAlias

rel_direction and dependent_part; select from it with
TableValuedAlias

select(list_navigable_roles()).
Source code in xmas_core/repositories/sql.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def list_navigable_roles() -> TableValuedAlias:
    """Build the ``list_navigable_roles`` call, which returns every configured navigable role.

    Returns:
        A derived table with one row per configured role and columns ``source_featuretype``,
        ``navigable_role``, ``target_featuretype``, ``appschema``, ``appschema_version``,
        ``rel_direction`` and ``dependent_part``; select from it with
        ``select(list_navigable_roles())``.
    """
    # The function projects the business columns of navigable_roles_config, dropping the
    # surrogate `id` and the `created_at` audit column that the ORM model also carries.
    role_columns = [
        column
        for column in NavigableRolesConfig.__table__.c
        if column.name not in ("id", "created_at")
    ]
    return _CoretableFunction("list_navigable_roles").table_valued(*role_columns)

list_top_level_featuretypes()

Build the list_top_level_featuretypes call.

Returns:

Type Description
TableValuedAlias

A derived table with one row per appschema/version and columns featuretype,
TableValuedAlias

appschema and appschema_version; select from it with
TableValuedAlias

select(list_top_level_featuretypes()).
Source code in xmas_core/repositories/sql.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
def list_top_level_featuretypes() -> TableValuedAlias:
    """Build the ``list_top_level_featuretypes`` call.

    Returns:
        A derived table with one row per appschema/version and columns ``featuretype``,
        ``appschema`` and ``appschema_version``; select from it with
        ``select(list_top_level_featuretypes())``.
    """
    return _CoretableFunction("list_top_level_featuretypes").table_valued(
        "featuretype",
        "appschema",
        "appschema_version",
    )

pg_array(values, type_)

Bind a Python sequence as a single PostgreSQL array parameter.

Keeps a statement's bind-parameter count independent of len(values): asyncpg refuses more than 32767 parameters per statement, and a bind list whose length varies also defeats the driver's prepared-statement cache, since it caches by SQL text.

Parameters:

Name Type Description Default
values Sequence[Any]

The values to bind; element bind processing runs through type_.

 required
type_ TypeEngine[Any]

The array's item type, e.g. Refs.base_id.type.

 required

Returns:

Type Description
Cast[Any]

A CAST(:param AS <type>[]) expression carrying exactly one bind parameter. The
Cast[Any]

cast is explicit because PostgreSQL cannot infer a polymorphic function's argument
Cast[Any]

type from an untyped parameter (see unnest_rows); SQLAlchemy's own array bind cast
Cast[Any]

may render a second, redundant ::<type>[] alongside it.
Source code in xmas_core/repositories/sql.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def pg_array(values: Sequence[Any], type_: TypeEngine[Any]) -> Cast[Any]:
    """Bind a Python sequence as a single PostgreSQL array parameter.

    Keeps a statement's bind-parameter count independent of ``len(values)``: asyncpg refuses
    more than 32767 parameters per statement, and a bind list whose length varies also defeats
    the driver's prepared-statement cache, since it caches by SQL text.

    Args:
        values: The values to bind; element bind processing runs through ``type_``.
        type_: The array's item type, e.g. ``Refs.base_id.type``.

    Returns:
        A ``CAST(:param AS <type>[])`` expression carrying exactly one bind parameter. The
        cast is explicit because PostgreSQL cannot infer a polymorphic function's argument
        type from an untyped parameter (see `unnest_rows`); SQLAlchemy's own array bind cast
        may render a second, redundant ``::<type>[]`` alongside it.
    """
    # dimensions=1 is what every array here is, and saying so halves the copying: with the
    # dimensions unset, `ARRAY._apply_item_processor` cannot know the sequence is flat, so it
    # materialises `list(values)` just to look at `values[0]` before building the list it
    # returns. It does not change the rendered SQL or the bound value.
    array_type = ARRAY(type_, dimensions=1)
    return cast(
        literal(values, array_type),
        array_type,
    )

unnest_rows(rows, columns, *, types=None)

Expose row dicts as a table-valued unnest over one array parameter per column.

The PostgreSQL counterpart to an inline VALUES list: instead of one bind parameter per cell it binds one array per column, so the parameter count stays constant (see pg_array).

Parameters:

Name Type Description Default
rows Sequence[dict[str, Any]]

Row dicts keyed by column name; every column in columns must be present.

 required
columns Sequence[Column[Any]]

The columns to emit, in order.

 required
types Mapping[str, TypeEngine[Any]] | None

Array item types overriding a column's own type, keyed by column name. Needed where the column type is not what the driver can encode an array of — geometry columns travel as TEXT[] of eWKT and are converted in the select list.

 None

Returns:

Type Description
TableValuedAlias

A derived table aliased with columns' names, addressable via .c.<name>.
Source code in xmas_core/repositories/sql.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def unnest_rows(
    rows: Sequence[dict[str, Any]],
    columns: Sequence[Column[Any]],
    *,
    types: Mapping[str, TypeEngine[Any]] | None = None,
) -> TableValuedAlias:
    """Expose row dicts as a table-valued ``unnest`` over one array parameter per column.

    The PostgreSQL counterpart to an inline ``VALUES`` list: instead of one bind parameter per
    cell it binds one array per column, so the parameter count stays constant (see `pg_array`).

    Args:
        rows: Row dicts keyed by column name; every column in ``columns`` must be present.
        columns: The columns to emit, in order.
        types: Array item types overriding a column's own type, keyed by column name. Needed
            where the column type is not what the driver can encode an array of — geometry
            columns travel as ``TEXT[]`` of eWKT and are converted in the select list.

    Returns:
        A derived table aliased with ``columns``' names, addressable via ``.c.<name>``.
    """
    types = types or {}
    return (
        func.unnest(
            *[
                pg_array([r[c.name] for r in rows], types.get(c.name, c.type))
                for c in columns
            ]
        )
        .table_valued(*[c.name for c in columns])
        # Multi-argument unnest names every output column "unnest", so the derived column
        # alias list is what makes `.c.<name>` resolvable.
        .render_derived()
    )