pylingEYE N3 Reasoner

Notebooks PyPI GitHub

Pyling notebook 05

Can Alice read resource X right now?

An access request rarely depends on identity alone. A policy might grant the action only on particular days, during particular hours, for a particular resource. The answer must combine three things:

1. the ODRL policy, which describes the rule;
2. the request, which describes the intended action;
3. the state of the world, which supplies facts such as the current time.

That is the evaluation model demonstrated by FORCE: Framework for ODRL Rule Compliance through Evaluation. FORCE returns a compliance report that says which requested rule was attempted and whether the policy rule is active.

We will reconstruct the decisive reasoning path through the ODRL Test Suite's most demanding example, testcase-062-big-policy: Alice may read resource X from 09:00 until 17:00 on every weekday of 2024. The request arrives at 11:20 on Monday, 12 February.

1. Why this example is genuinely complex

The policy does not say "weekday" through a procedural calendar function. It expands the year into an ODRL logical constraint:

• one top-level odrl:or;
• 262 weekday branches in the leap year 2024;
• one odrl:and per branch;
• two atomic date-time constraints per day: after 09:00 and before 17:00.

That gives 524 atomic constraints. Only one complete branch must be satisfied, but the evaluator must avoid combining the lower bound of one day with the upper bound of another.

Generating this repetitive RDF in Python keeps the notebook readable while preserving the structure and semantics of the published FORCE fixture.

from datetime import datetime, timedelta, timezone

from rdflib import Graph, Literal as RDFLiteral, Namespace, RDF
from rdflib.namespace import DCTERMS, XSD
from pyling import Literal, XSD_NS, reason_stream, register_builtin, unregister_builtin

ODRL = Namespace("http://www.w3.org/ns/odrl/2/")
EX = Namespace("http://example.org/")
TEMP = Namespace("http://example.com/request/")
REPORT = Namespace("https://w3id.org/force/compliance-report#")
FORCE = Namespace("https://w3id.org/force/notebook#")


def xsd_datetime(value):
    lexical = value.isoformat(timespec="milliseconds").replace("+00:00", "Z")
    return RDFLiteral(lexical, datatype=XSD.dateTime)


policy_graph = Graph()
for prefix, namespace in (
    ("odrl", ODRL),
    ("ex", EX),
    ("temp", TEMP),
    ("report", REPORT),
    ("force", FORCE),
    ("dct", DCTERMS),
):
    policy_graph.bind(prefix, namespace)

policy = EX.officeHoursPolicy
permission = EX.officeHoursPermission
top_constraint = EX.weekdayWindows
request = EX.aliceReadRequest
requested_permission = EX.requestedPermission

policy_graph.add((policy, RDF.type, ODRL.Set))
policy_graph.add((policy, DCTERMS.description, RDFLiteral(
    "During 09:00-17:00 on weekdays of 2024, Alice may read resource X."
)))
policy_graph.add((policy, ODRL.permission, permission))
policy_graph.add((permission, RDF.type, ODRL.Permission))
for predicate, value in (
    (ODRL.assignee, EX.alice),
    (ODRL.action, ODRL.read),
    (ODRL.target, EX.x),
    (ODRL.constraint, top_constraint),
):
    policy_graph.add((permission, predicate, value))
policy_graph.add((top_constraint, RDF.type, ODRL.LogicalConstraint))

policy_graph.add((request, RDF.type, ODRL.Request))
policy_graph.add((request, ODRL.permission, requested_permission))
policy_graph.add((requested_permission, RDF.type, ODRL.Permission))
for predicate, value in (
    (ODRL.assignee, EX.alice),
    (ODRL.action, ODRL.read),
    (ODRL.target, EX.x),
):
    policy_graph.add((requested_permission, predicate, value))

current_time = datetime(2024, 2, 12, 11, 20, 10, 999000, tzinfo=timezone.utc)
policy_graph.add((TEMP.currentTime, DCTERMS.issued, xsd_datetime(current_time)))

day = datetime(2024, 1, 1, tzinfo=timezone.utc)
weekday_count = 0
while day.year == 2024:
    if day.weekday() < 5:
        weekday_count += 1
        date = day.date().isoformat()
        branch = EX[f"window-{date}"]
        lower = EX[f"after-{date}-09"]
        upper = EX[f"before-{date}-17"]

        policy_graph.add((top_constraint, ODRL["or"], branch))
        policy_graph.add((branch, RDF.type, ODRL.LogicalConstraint))
        policy_graph.add((branch, ODRL["and"], lower))
        policy_graph.add((branch, ODRL["and"], upper))

        for node, operator, hour in (
            (lower, ODRL.gt, 9),
            (upper, ODRL.lt, 17),
        ):
            policy_graph.add((node, RDF.type, ODRL.Constraint))
            policy_graph.add((node, ODRL.leftOperand, ODRL.dateTime))
            policy_graph.add((node, ODRL.operator, operator))
            policy_graph.add((node, ODRL.rightOperand, xsd_datetime(day + timedelta(hours=hour))))
    day += timedelta(days=1)

print(f"{weekday_count} weekday branches")
print(f"{weekday_count * 2} atomic date-time constraints")
print(f"{len(policy_graph):,} input triples in policy, request, and world state")
assert weekday_count == 262

262 weekday branches
524 atomic date-time constraints
3,160 input triples in policy, request, and world state

2. Zoom in before evaluating the whole year

The February 12 branch shows the shape repeated for every weekday. The top or points to a daily window; that window requires both its lower and upper bounds. Keeping those bounds under the same and node is what prevents a false match across dates.

sample_branch = EX["window-2024-02-12"]
print("Top constraint OR branch:", sample_branch.n3(policy_graph.namespace_manager))
for constraint in policy_graph.objects(sample_branch, ODRL["and"]):
    operator = policy_graph.value(constraint, ODRL.operator)
    boundary = policy_graph.value(constraint, ODRL.rightOperand)
    print(
        "  AND",
        operator.n3(policy_graph.namespace_manager),
        boundary.n3(policy_graph.namespace_manager),
    )
print("Current time:", policy_graph.value(TEMP.currentTime, DCTERMS.issued))

Top constraint OR branch: ex:window-2024-02-12
  AND odrl:gt "2024-02-12T09:00:00+00:00"^^xsd:dateTime
  AND odrl:lt "2024-02-12T17:00:00+00:00"^^xsd:dateTime
Current time: 2024-02-12T11:20:10.999000+00:00

3. Connect the state of the world to N3

FORCE resolves ODRL's dateTime left operand from the state of the world. In this Python notebook, two small custom built-ins give N3 timezone-aware date-time comparison while leaving the policy logic in RDF.

The built-ins do not know who Alice is, what she wants, or which hours are permitted. They only answer the reusable comparison questions "is this instant after that one?" and "is it before that one?"

def date_time_value(term):
    if not isinstance(term, Literal) or term.datatype != XSD_NS + "dateTime":
        return None
    return datetime.fromisoformat(term.lexical.replace("Z", "+00:00"))


def date_time_comparison(operator):
    def handler(context):
        left = context.engine.apply_subst(context.goal.s, context.subst)
        right = context.engine.apply_subst(context.goal.o, context.subst)
        left_value = date_time_value(left)
        right_value = date_time_value(right)
        if left_value is None or right_value is None:
            return []
        return [context.subst] if operator(left_value, right_value) else []

    return handler


greater_than_iri = str(FORCE.dateTimeGreaterThan)
less_than_iri = str(FORCE.dateTimeLessThan)
register_builtin(greater_than_iri, date_time_comparison(lambda left, right: left > right))
register_builtin(less_than_iri, date_time_comparison(lambda left, right: left < right))

<function __main__.date_time_comparison.<locals>.handler(context)>

4. Follow the compliance decision one layer at a time

The rules mirror the shape of the policy:

1. resolve and test each atomic date-time constraint;
2. satisfy a daily and only when both that day's bounds hold;
3. satisfy the top or when any complete daily branch holds;
4. match assignee, action, and target between policy and request;
5. emit a FORCE compliance report with an active permission report.

The final rule cannot fire merely because Alice asked to read X. The matching policy constraint must also have reached report:Satisfied.

evaluation_rules = """
@prefix odrl: <http://www.w3.org/ns/odrl/2/> .
@prefix dct: <http://purl.org/dc/terms/> .
@prefix temp: <http://example.com/request/> .
@prefix report: <https://w3id.org/force/compliance-report#> .
@prefix force: <https://w3id.org/force/notebook#> .

{
  ?constraint odrl:leftOperand odrl:dateTime ;
              odrl:operator odrl:gt ;
              odrl:rightOperand ?boundary .
  temp:currentTime dct:issued ?now .
  ?now force:dateTimeGreaterThan ?boundary .
} => {
  ?constraint report:satisfactionState report:Satisfied .
} .

{
  ?constraint odrl:leftOperand odrl:dateTime ;
              odrl:operator odrl:lt ;
              odrl:rightOperand ?boundary .
  temp:currentTime dct:issued ?now .
  ?now force:dateTimeLessThan ?boundary .
} => {
  ?constraint report:satisfactionState report:Satisfied .
} .

{
  ?branch odrl:and ?lower ;
          odrl:and ?upper .
  ?lower odrl:operator odrl:gt ;
         report:satisfactionState report:Satisfied .
  ?upper odrl:operator odrl:lt ;
         report:satisfactionState report:Satisfied .
} => {
  ?branch report:satisfactionState report:Satisfied .
} .

{
  ?top odrl:or ?branch .
  ?branch report:satisfactionState report:Satisfied .
} => {
  ?top report:satisfactionState report:Satisfied .
} .

{
  ?policy a odrl:Set ;
          odrl:permission ?permission .
  ?permission odrl:assignee ?assignee ;
              odrl:action ?action ;
              odrl:target ?target ;
              odrl:constraint ?constraint .

  ?request a odrl:Request ;
           odrl:permission ?requestedPermission .
  ?requestedPermission odrl:assignee ?assignee ;
                       odrl:action ?action ;
                       odrl:target ?target .

  ?constraint report:satisfactionState report:Satisfied .
} => {
  force:report a report:PolicyReport ;
      report:policy ?policy ;
      report:policyRequest ?request ;
      report:ruleReport force:permissionReport .

  force:permissionReport a report:PermissionReport ;
      report:rule ?permission ;
      report:ruleRequest ?requestedPermission ;
      report:attemptState report:Attempted ;
      report:activationState report:Active .
} .
"""

5. Run the evaluation and demand an auditable answer

The custom built-ins are registered only for this reasoning run and removed afterwards. We retain input facts in the closure so the report can be traced back through the satisfied logical branch to its original bounds.

try:
    result = reason_stream(
        {"sources": [policy_graph.serialize(format="turtle"), evaluation_rules]},
        include_input_facts_in_closure=True,
        max_iterations=20,
    )
finally:
    unregister_builtin(greater_than_iri)
    unregister_builtin(less_than_iri)

closure = result.as_rdflib_graph(include_input_facts=True)
active_statement = (
    FORCE.permissionReport,
    REPORT.activationState,
    REPORT.Active,
)
assert active_statement in closure
print("Compliance decision: the requested permission is ACTIVE")
print(f"Derived {len(result.derived):,} triples")

Compliance decision: the requested permission is ACTIVE
Derived 274 triples

6. Find the evidence that made the report active

Many individual "after 09:00" or "before 17:00" tests are true in isolation. A daily branch is stronger: both bounds for the same date must be true. Exactly one such branch should satisfy the top-level or at the chosen instant.

satisfied_branches = [
    branch
    for branch in closure.subjects(REPORT.satisfactionState, REPORT.Satisfied)
    if (top_constraint, ODRL["or"], branch) in closure
]
assert satisfied_branches == [EX["window-2024-02-12"]]

branch = satisfied_branches[0]
bounds = []
for constraint in closure.objects(branch, ODRL["and"]):
    bounds.append((
        closure.value(constraint, ODRL.operator),
        closure.value(constraint, ODRL.rightOperand),
    ))

print("Satisfied daily branch:", branch.n3(closure.namespace_manager))
for operator, boundary in sorted(bounds, key=lambda item: str(item[0])):
    print(" ", operator.n3(closure.namespace_manager), boundary)

Satisfied daily branch: ex:window-2024-02-12
  odrl:gt 2024-02-12T09:00:00+00:00
  odrl:lt 2024-02-12T17:00:00+00:00

7. Read the report as RDF, not as a boolean

A boolean would answer today's question and discard the reasoning context. The FORCE report vocabulary preserves the policy, request, attempted permission, and activation state as linked resources.

report_graph = Graph()
for subject in (FORCE["report"], FORCE.permissionReport):
    for triple in closure.triples((subject, None, None)):
        report_graph.add(triple)
for prefix, namespace in (("report", REPORT), ("force", FORCE), ("ex", EX)):
    report_graph.bind(prefix, namespace)

print(report_graph.serialize(format="turtle"))

@prefix ex: <http://example.org/> .
@prefix force: <https://w3id.org/force/notebook#> .
@prefix report: <https://w3id.org/force/compliance-report#> .

force:report a report:PolicyReport ;
    report:policy ex:officeHoursPolicy ;
    report:policyRequest ex:aliceReadRequest ;
    report:ruleReport force:permissionReport .

force:permissionReport a report:PermissionReport ;
    report:activationState report:Active ;
    report:attemptState report:Attempted ;
    report:rule ex:officeHoursPermission ;
    report:ruleRequest ex:requestedPermission .

What this tells us

The request matched the policy's assignee (ex:alice), action (odrl:read), and target (ex:x). The world-state time, 11:20 UTC on Monday 12 February, was later than that day's 09:00 lower bound and earlier than its 17:00 upper bound. That daily and satisfied the year-wide or, so the permission report became report:Active.

This notebook intentionally isolates the successful decision path so it can be read end to end. The production ODRL Evaluator runs a multi-stage FORCE ruleset and emits richer premise reports, including unsatisfied constraints and inactive rules. The input structure, logical constraint semantics, request matching, and report vocabulary here follow the published testcase-062-big-policy; the smaller rule profile is educational rather than a claim of full FORCE conformance.

The recurring pattern across all five notebooks is now visible: keep facts, reusable semantics, runtime context, and derived evidence separate. That separation is what makes a reasoning result useful beyond the moment in which it was computed.

Python N3 reasoning in the EYE ecosystem. All notebooks