Python

Complete examples

Examples in the unified Python binding.

The Python complete examples reference continues the Python overview.

Document-access example

The following three fragments build one program, in order. They use only methods already implemented in Python.

Step 1 — Declare the vocabulary

Our document-access example declares every predicate before loading the policy. owns, delegated, blocked and user are runtime inputs. can_read is an intermediate relation; allow and has_any_document are queryable outputs.

Start documents_next.py with the imports and predicate declarations:

CODE

from maelys_datalog import Engine, Predicate, PRED_EDB, PRED_IDB, PRED_QUERY

predicates = [
    Predicate("user", 1, PRED_EDB),
    Predicate("owns", 2, PRED_EDB),
    Predicate("delegated", 2, PRED_EDB),
    Predicate("blocked", 1, PRED_EDB),
    Predicate("can_read", 2, PRED_IDB),
    Predicate("has_any_document", 1, PRED_IDB | PRED_QUERY),
    Predicate("allow", 2, PRED_IDB | PRED_QUERY),
]

Each Predicate supplies a name, an arity (number of arguments), and kind flags. PRED_EDB marks input facts; PRED_IDB marks derived facts. Add PRED_QUERY to a derived predicate when the application needs to query it. The optional constructors above express these same declarations more briefly.

This list describes the vocabulary; Step 3 registers it with the engine. A domain is process-wide, not private to an Engine. Reuse a stable definition instead of inventing a new name for every request.

Step 2 — Express the policy

DATALOG

can_read(User, Doc) :-
    owns(User, Doc) or delegated(User, Doc),
    not(blocked(User)).

has_any_document(User) :- owns(User, _).

allow(User, Doc) :-
    user(User),
    can_read(User, Doc).

The or offers two positive access paths; not(blocked(User)) filters both. The wildcard _ means that any document is enough for has_any_document. All of these predicate names were declared in Step 1; the language does not declare them implicitly.

Append the same rules as a Python multiline string named policy. The triple quotes delimit the Python string; its contents are Datalog, not Python. This step defines the source text — loading happens in Step 3.

CODE

policy = """
can_read(User, Doc) :-
    owns(User, Doc) or delegated(User, Doc),
    not(blocked(User)).
has_any_document(User) :- owns(User, _).
allow(User, Doc) :- user(User), can_read(User, Doc).
"""

Step 3 — Add facts, solve and query

Append this block after predicates and policy. It creates the engine, registers the vocabulary, loads the policy, then evaluates one batch of facts. There is no need to redefine either variable.

CODE

with Engine() as engine:
    engine.register_domain("documents_next", predicates)
    ruleset = engine.load_inline_ruleset("documents_next", "documents.main", policy)

    edb = ruleset.edb()
    edb.add_fact("user", ["alice"])
    edb.add_facts([
        ("user", ["bob"]),
        ("user", ["mallory"]),
        ("owns", ["alice", "roadmap.pdf"]),
        ("delegated", ["bob", "roadmap.pdf"]),
        ("owns", ["mallory", "roadmap.pdf"]),
        ("blocked", ["mallory"]),
    ])
    result = ruleset.solve(edb)
    for user in ("alice", "bob", "mallory"):
        print(user, result.contains_fact("allow", [user, "roadmap.pdf"]))
    print("allow:", sorted(result.enumerate_predicate_facts("allow", 2)))
    print("derived:", result.derived_fact_count())

register_domain() registers the declarations from Step 1. load_inline_ruleset() compiles the source from Step 2 against that domain. The additions buffer the runtime facts; solve() submits them to the native engine, and the final calls inspect its result. The with block closes the engine and its children when it exits.

Complete program

The three fragments above, in order, form this standalone documents_next.py. Copy either the three fragments or the complete program below — not both.

CODE

from maelys_datalog import Engine, Predicate, PRED_EDB, PRED_IDB, PRED_QUERY

predicates = [
    Predicate("user", 1, PRED_EDB),
    Predicate("owns", 2, PRED_EDB),
    Predicate("delegated", 2, PRED_EDB),
    Predicate("blocked", 1, PRED_EDB),
    Predicate("can_read", 2, PRED_IDB),
    Predicate("has_any_document", 1, PRED_IDB | PRED_QUERY),
    Predicate("allow", 2, PRED_IDB | PRED_QUERY),
]
policy = """
can_read(User, Doc) :-
    owns(User, Doc) or delegated(User, Doc),
    not(blocked(User)).
has_any_document(User) :- owns(User, _).
allow(User, Doc) :- user(User), can_read(User, Doc).
"""

with Engine() as engine:
    engine.register_domain("documents_next", predicates)
    ruleset = engine.load_inline_ruleset("documents_next", "documents.main", policy)

    edb = ruleset.edb()
    edb.add_fact("user", ["alice"])
    edb.add_facts([
        ("user", ["bob"]),
        ("user", ["mallory"]),
        ("owns", ["alice", "roadmap.pdf"]),
        ("delegated", ["bob", "roadmap.pdf"]),
        ("owns", ["mallory", "roadmap.pdf"]),
        ("blocked", ["mallory"]),
    ])
    result = ruleset.solve(edb)
    for user in ("alice", "bob", "mallory"):
        print(user, result.contains_fact("allow", [user, "roadmap.pdf"]))
    print("allow:", sorted(result.enumerate_predicate_facts("allow", 2)))
    print("derived:", result.derived_fact_count())

Run it from the repository root:

CODE

PYTHONPATH=bindings/python python documents_next.py

Expected output:

CODE

alice True
bob True
mallory False
allow: [('alice', 'roadmap.pdf'), ('bob', 'roadmap.pdf')]
derived: 6

Alice owns the document. Bob has delegated access. Mallory owns it too, but is blocked. The count of six includes two can_read facts, two has_any_document facts and two allow facts — not just the two rows returned by the allow enumeration.