`djangordf.schema`¶

Django-migrations-style framework for evolving the RDF schema and data in the triple store. Define one Migration subclass per change under the configured DJANGORDF_MIGRATIONS_MODULE, then run python manage.py migrate_rdf.

Django-style migrations for the RDF schema.

Define a Migration subclass per change, list operations from djangordf.schema.operations, point at any dependencies by migration name, and run python manage.py migrate_rdf to apply.

The applied-state recorder lives in the triple store itself (urn:djangordf:migrations) so the next run knows what is already in place.

class djangordf.schema.AddPropertyDeclaration(predicate, *, kind, domain=None, range=None, graph=None)[source]¶

Bases: Operation

Declare a predicate in the ontology graph.

kind is either "datatype" or "object". Domain and range are optional; when present they are emitted as rdfs:domain / rdfs:range triples on the predicate IRI.

Parameters:

predicate (str | URIRef)
kind (str)
domain (str | URIRef | None)
range (str | URIRef | None)
graph (str | URIRef | None)

apply(backend)[source]¶

Return type:: None

class djangordf.schema.CreateClass(class_iri, *, label=None, comment=None, graph=None)[source]¶

Bases: Operation

Append an owl:Class declaration to the ontology graph.

Optionally writes rdfs:label and rdfs:comment annotations alongside.

Parameters:

class_iri (str | URIRef)
label (str | None)
comment (str | None)
graph (str | URIRef | None)

apply(backend)[source]¶

Return type:: None

class djangordf.schema.DeleteClass(class_iri, *, graph=None)[source]¶

Bases: Operation

Strip every triple whose subject is the class IRI from the ontology graph.

Parameters:

class_iri (str | URIRef)
graph (str | URIRef | None)

apply(backend)[source]¶

Return type:: None

class djangordf.schema.Executor(backend=None, module_path=None)[source]¶

Bases: object

Discover Migration subclasses, sort them topologically, and apply any that the recorder has not seen yet.

Parameters:: module_path (str | None)

discover()[source]¶

Import every *.py under the configured module, collect Migration subclasses, and return one instance per class sorted lexically by name.

Return type:: List[Migration]

migrate()[source]¶

Apply every pending migration in dependency order and return the list of newly-applied migrations.

Return type:: List[Migration]

plan()[source]¶

Return (applied, pending) lists in execution order.

Return type:: Tuple[List[Migration], List[Migration]]

class djangordf.schema.Migration[source]¶

Bases: object

One forward-only RDF schema or data change.

apply(backend)[source]¶

Run every operation in declaration order against backend.

Return type:: None

dependencies: List[str] = []¶

name: Optional[str] = None¶

operations: List[Operation] = []¶

class djangordf.schema.MigrationRecorder(backend)[source]¶

Bases: object

Tiny wrapper around backend that tracks applied migrations in a dedicated named graph.

applied()[source]¶

Return the set of migration names already recorded.

Return type:: Set[str]

forget(name)[source]¶

Strip the record for name from the graph (used in tests; no production command depends on this).

Return type:: None
Parameters:: name (str)

forget_all(names)[source]¶

Return type:: None
Parameters:: names (Iterable[str])

record(name, *, when)[source]¶

Persist that name has been applied at when (an ISO-8601 timestamp). when is provided by the caller so the executor can stamp every applied migration with the same run-wide timestamp if desired.

Return type:

None

Parameters:

name (str)
when (str)

class djangordf.schema.Operation[source]¶

Bases: object

Marker base class for migration operations.

apply(backend)[source]¶

Return type:: None

class djangordf.schema.RenamePredicate(old, new, *, graph=None)[source]¶

Bases: Operation

Rewrite every (?s, old, ?o) triple to (?s, new, ?o).

Runs across the data graph by default; an explicit graph= targets a specific named graph. Useful when the user-facing name of a property changes but the data should remain.

Parameters:

old (str | URIRef)
new (str | URIRef)
graph (str | URIRef | None)

apply(backend)[source]¶

Return type:: None

class djangordf.schema.RunSPARQL(sparql)[source]¶

Bases: Operation

Run an arbitrary SPARQL UPDATE statement against the backend.

The escape hatch for transformations that do not fit any of the higher-level operations.

Parameters:: sparql (str)

apply(backend)[source]¶

Return type:: None

The Migration base class.

Subclass it in a per-change file, list operations, and reference any prior migrations by name through dependencies:

class Migration(Migration):
    name = "0002_rename_pref"
    dependencies = ["0001_initial"]
    operations = [
        RenamePredicate(
            old=URIRef("http://example.org/oldPred"),
            new=URIRef("http://example.org/newPred"),
        ),
    ]

The name field defaults to the module name if not set.

class djangordf.schema.base.Migration[source]¶

Bases: object

One forward-only RDF schema or data change.

name: Optional[str] = None¶

dependencies: List[str] = []¶

operations: List[Operation] = []¶

apply(backend)[source]¶

Run every operation in declaration order against backend.

Return type:: None

Operation types that can appear in a Migration.operations list.

All operations carry their own apply(backend) method. Operations are forward-only in this release; rollback support is a separate ticket.

class djangordf.schema.operations.Operation[source]¶

Bases: object

Marker base class for migration operations.

apply(backend)[source]¶

Return type:: None

class djangordf.schema.operations.RunSPARQL(sparql)[source]¶

Bases: Operation

Run an arbitrary SPARQL UPDATE statement against the backend.

The escape hatch for transformations that do not fit any of the higher-level operations.

Parameters:: sparql (str)

apply(backend)[source]¶

Return type:: None

class djangordf.schema.operations.CreateClass(class_iri, *, label=None, comment=None, graph=None)[source]¶

Bases: Operation

Append an owl:Class declaration to the ontology graph.

Optionally writes rdfs:label and rdfs:comment annotations alongside.

Parameters:

class_iri (str | URIRef)
label (str | None)
comment (str | None)
graph (str | URIRef | None)

apply(backend)[source]¶

Return type:: None

class djangordf.schema.operations.DeleteClass(class_iri, *, graph=None)[source]¶

Bases: Operation

Strip every triple whose subject is the class IRI from the ontology graph.

Parameters:

class_iri (str | URIRef)
graph (str | URIRef | None)

apply(backend)[source]¶

Return type:: None

class djangordf.schema.operations.AddPropertyDeclaration(predicate, *, kind, domain=None, range=None, graph=None)[source]¶

Bases: Operation

Declare a predicate in the ontology graph.

kind is either "datatype" or "object". Domain and range are optional; when present they are emitted as rdfs:domain / rdfs:range triples on the predicate IRI.

Parameters:

predicate (str | URIRef)
kind (str)
domain (str | URIRef | None)
range (str | URIRef | None)
graph (str | URIRef | None)

apply(backend)[source]¶

Return type:: None

class djangordf.schema.operations.RenamePredicate(old, new, *, graph=None)[source]¶

Bases: Operation

Rewrite every (?s, old, ?o) triple to (?s, new, ?o).

Runs across the data graph by default; an explicit graph= targets a specific named graph. Useful when the user-facing name of a property changes but the data should remain.

Parameters:

old (str | URIRef)
new (str | URIRef)
graph (str | URIRef | None)

apply(backend)[source]¶

Return type:: None

Discover, sort and apply RDF migrations.

class djangordf.schema.executor.Executor(backend=None, module_path=None)[source]¶

Bases: object

Discover Migration subclasses, sort them topologically, and apply any that the recorder has not seen yet.

Parameters:: module_path (str | None)

discover()[source]¶

Import every *.py under the configured module, collect Migration subclasses, and return one instance per class sorted lexically by name.

Return type:: List[Migration]

plan()[source]¶

Return (applied, pending) lists in execution order.

Return type:: Tuple[List[Migration], List[Migration]]

migrate()[source]¶

Apply every pending migration in dependency order and return the list of newly-applied migrations.

Return type:: List[Migration]

Track applied migrations as triples in the configured backend.

class djangordf.schema.recorder.MigrationRecorder(backend)[source]¶

Bases: object

Tiny wrapper around backend that tracks applied migrations in a dedicated named graph.

applied()[source]¶

Return the set of migration names already recorded.

Return type:: Set[str]

record(name, *, when)[source]¶

Persist that name has been applied at when (an ISO-8601 timestamp). when is provided by the caller so the executor can stamp every applied migration with the same run-wide timestamp if desired.

Return type:

None

Parameters:

name (str)
when (str)

forget(name)[source]¶

Strip the record for name from the graph (used in tests; no production command depends on this).

Return type:: None
Parameters:: name (str)

forget_all(names)[source]¶

Return type:: None
Parameters:: names (Iterable[str])

djangordf.schema¶

`djangordf.schema`¶