Index

shotgrid_casbin_adapter

ShotGrid Casbin Adapter.

A Casbin policy adapter that enables loading and saving access control policies from/to Autodesk ShotGrid (formerly Shotgun).

Adapter

Adapter(sg: Shotgun | None = None, base_url: str | None = None, script_name: str | None = None, api_key: str | None = None, entity_type: str | None = None, project_id: int | None = None, filtered: bool = False)

Bases: Adapter, UpdateAdapter

Casbin policy adapter backed by Autodesk ShotGrid.

This adapter maps Casbin policy rules to ShotGrid custom entity records. Each record contains ptype and v0-v5 fields representing one policy line.

When project_id is provided, all operations are scoped to that project: queries include a project filter, and created entities are linked to the project. This enables per-project policy isolation within a single ShotGrid site. When project_id is None, the adapter operates at site level.

ShotGrid's delete() operation retires entities rather than destroying them, and find() excludes retired records by default. This provides natural soft-delete behavior without additional configuration.

PARAMETER	DESCRIPTION
sg	An existing shotgun_api3.Shotgun instance. If provided, base_url, script_name, and api_key are ignored. TYPE: Shotgun | None DEFAULT: None
base_url	ShotGrid server URL. Falls back to SHOTGRID_URL env var. TYPE: str | None DEFAULT: None
script_name	ShotGrid script name. Falls back to SHOTGRID_SCRIPT_NAME env var. TYPE: str | None DEFAULT: None
api_key	ShotGrid API key. Falls back to SHOTGRID_API_KEY env var. TYPE: str | None DEFAULT: None
entity_type	ShotGrid entity type for storing rules. Falls back to SHOTGRID_ENTITY_TYPE env var, then DEFAULT_ENTITY_TYPE. TYPE: str | None DEFAULT: None
project_id	ShotGrid project ID for scoping operations. Falls back to SHOTGRID_PROJECT_ID env var. When None, operations are site-wide. TYPE: int | None DEFAULT: None
filtered	Whether this adapter supports filtered policy loading. TYPE: bool DEFAULT: False

RAISES	DESCRIPTION
ValueError	If neither sg nor sufficient connection parameters are provided.

Initialize the ShotGrid Casbin adapter.

PARAMETER	DESCRIPTION
sg	An existing shotgun_api3.Shotgun instance. If provided, base_url, script_name, and api_key are ignored. TYPE: Shotgun | None DEFAULT: None
base_url	ShotGrid server URL. Falls back to SHOTGRID_URL env var. TYPE: str | None DEFAULT: None
script_name	ShotGrid script name. Falls back to SHOTGRID_SCRIPT_NAME env var. TYPE: str | None DEFAULT: None
api_key	ShotGrid API key. Falls back to SHOTGRID_API_KEY env var. TYPE: str | None DEFAULT: None
entity_type	ShotGrid entity type for storing rules. Falls back to SHOTGRID_ENTITY_TYPE env var, then DEFAULT_ENTITY_TYPE. TYPE: str | None DEFAULT: None
project_id	ShotGrid project ID for scoping operations. Falls back to SHOTGRID_PROJECT_ID env var. When None, operations are site-wide. TYPE: int | None DEFAULT: None
filtered	Whether this adapter supports filtered policy loading. TYPE: bool DEFAULT: False

Source code in shotgrid_casbin_adapter/core.py

def __init__(
    self,
    sg: Shotgun | None = None,
    base_url: str | None = None,
    script_name: str | None = None,
    api_key: str | None = None,
    entity_type: str | None = None,
    project_id: int | None = None,
    filtered: bool = False,
) -> None:
    """Initialize the ShotGrid Casbin adapter.

    Args:
        sg: An existing ``shotgun_api3.Shotgun`` instance. If provided,
            ``base_url``, ``script_name``, and ``api_key`` are ignored.
        base_url: ShotGrid server URL. Falls back to ``SHOTGRID_URL`` env var.
        script_name: ShotGrid script name. Falls back to ``SHOTGRID_SCRIPT_NAME`` env var.
        api_key: ShotGrid API key. Falls back to ``SHOTGRID_API_KEY`` env var.
        entity_type: ShotGrid entity type for storing rules.
            Falls back to ``SHOTGRID_ENTITY_TYPE`` env var, then ``DEFAULT_ENTITY_TYPE``.
        project_id: ShotGrid project ID for scoping operations.
            Falls back to ``SHOTGRID_PROJECT_ID`` env var. When ``None``,
            operations are site-wide.
        filtered: Whether this adapter supports filtered policy loading.
    """
    self._entity_type: str = entity_type or os.environ.get(SHOTGRID_ENTITY_TYPE) or DEFAULT_ENTITY_TYPE
    self._project_id: int | None = (
        project_id if project_id is not None else (int(v) if (v := os.environ.get(SHOTGRID_PROJECT_ID)) else None)
    )
    self._filtered: bool = filtered
    self._sg: Shotgun = sg or _connect_sg(base_url=base_url, script_name=script_name, api_key=api_key)

entity_type property

entity_type: str

The ShotGrid entity type used for storing Casbin rules.

project_id property

project_id: int | None

The ShotGrid project ID for scoping operations, or None for site-wide.

sg property

sg: Shotgun

The underlying shotgun_api3.Shotgun connection instance.

add_policies

add_policies(sec: str, ptype: str, rules: list[list[str]]) -> None

Add multiple policy rules to ShotGrid via batch operation.

PARAMETER	DESCRIPTION
sec	The policy section ("p" or "g"). TYPE: str
ptype	The policy type identifier. TYPE: str
rules	A list of policy rule value lists. TYPE: list[list[str]]

Source code in shotgrid_casbin_adapter/core.py

def add_policies(self, sec: str, ptype: str, rules: list[list[str]]) -> None:
    """Add multiple policy rules to ShotGrid via batch operation.

    Args:
        sec: The policy section (``"p"`` or ``"g"``).
        ptype: The policy type identifier.
        rules: A list of policy rule value lists.
    """
    if not rules:
        return
    self._sg.batch([_build_create_request(self._entity_type, ptype, r, self._project_id) for r in rules])

add_policy

add_policy(sec: str, ptype: str, rule: list[str]) -> None

Add a policy rule to ShotGrid.

PARAMETER	DESCRIPTION
sec	The policy section ("p" or "g"). TYPE: str
ptype	The policy type identifier. TYPE: str
rule	The policy rule values. TYPE: list[str]

Source code in shotgrid_casbin_adapter/core.py

def add_policy(self, sec: str, ptype: str, rule: list[str]) -> None:
    """Add a policy rule to ShotGrid.

    Args:
        sec: The policy section (``"p"`` or ``"g"``).
        ptype: The policy type identifier.
        rule: The policy rule values.
    """
    self._sg.create(self._entity_type, _rule_to_dict(ptype, rule, self._project_id))

is_filtered

is_filtered() -> bool

Check whether the adapter is in filtered mode.

RETURNS	DESCRIPTION
bool	True if a filtered policy has been loaded, False otherwise.

Source code in shotgrid_casbin_adapter/core.py

def is_filtered(self) -> bool:
    """Check whether the adapter is in filtered mode.

    Returns:
        ``True`` if a filtered policy has been loaded, ``False`` otherwise.
    """
    return self._filtered

load_filtered_policy

load_filtered_policy(model: Model, filter: Filter) -> None

Load policy rules matching the filter from ShotGrid.

After loading, the adapter is marked as filtered (is_filtered() returns True).

PARAMETER	DESCRIPTION
model	The Casbin model to populate with matching policy rules. TYPE: Model
filter	A :class:Filter instance specifying field value constraints. TYPE: Filter

Source code in shotgrid_casbin_adapter/core.py

def load_filtered_policy(self, model: Model, filter: Filter) -> None:
    """Load policy rules matching the filter from ShotGrid.

    After loading, the adapter is marked as filtered (``is_filtered()``
    returns ``True``).

    Args:
        model: The Casbin model to populate with matching policy rules.
        filter: A :class:`Filter` instance specifying field value constraints.
    """
    sg_filters = _project_filter(self._project_id) + _build_sg_filters(filter)
    for entity in self._sg.find(self._entity_type, sg_filters, _FIELDS_WITH_ID):
        persist.load_policy_line(_entity_to_str(entity), model)
    self._filtered = True

load_policy

load_policy(model: Model) -> None

Load all policy rules from ShotGrid into the Casbin model.

Queries all non-retired entities of the configured type (optionally scoped to the configured project) and feeds each one into :func:casbin.persist.load_policy_line.

PARAMETER	DESCRIPTION
model	The Casbin model to populate with policy rules. TYPE: Model

Source code in shotgrid_casbin_adapter/core.py

def load_policy(self, model: Model) -> None:
    """Load all policy rules from ShotGrid into the Casbin model.

    Queries all non-retired entities of the configured type (optionally
    scoped to the configured project) and feeds each one into
    :func:`casbin.persist.load_policy_line`.

    Args:
        model: The Casbin model to populate with policy rules.
    """
    for entity in self._sg.find(self._entity_type, _project_filter(self._project_id), _FIELDS_WITH_ID):
        persist.load_policy_line(_entity_to_str(entity), model)

remove_filtered_policy

remove_filtered_policy(sec: str, ptype: str, field_index: int, *field_values: str) -> bool

Remove policy rules matching a field filter from ShotGrid.

PARAMETER	DESCRIPTION
sec	The policy section ("p" or "g"). TYPE: str
ptype	The policy type identifier. TYPE: str
field_index	The starting field index (0-5) for filtering. TYPE: int
*field_values	The field values to match starting at field_index. TYPE: str DEFAULT: ()

RETURNS	DESCRIPTION
bool	True if at least one rule was removed, False otherwise.

Source code in shotgrid_casbin_adapter/core.py

def remove_filtered_policy(self, sec: str, ptype: str, field_index: int, *field_values: str) -> bool:
    """Remove policy rules matching a field filter from ShotGrid.

    Args:
        sec: The policy section (``"p"`` or ``"g"``).
        ptype: The policy type identifier.
        field_index: The starting field index (0-5) for filtering.
        *field_values: The field values to match starting at ``field_index``.

    Returns:
        ``True`` if at least one rule was removed, ``False`` otherwise.
    """
    if not (0 <= field_index <= 5) or not (1 <= field_index + len(field_values) <= 6):
        return False

    sg_filters: list[list[Any]] = [*_project_filter(self._project_id), [CASBIN_FIELDS[0], "is", ptype]]
    for i, v in enumerate(field_values):
        if v != "":
            sg_filters.append([CASBIN_FIELDS[field_index + i + 1], "is", v])

    entities = self._sg.find(self._entity_type, sg_filters, ["id"])
    if not entities:
        return False
    _batch_delete(self._sg, self._entity_type, entities)
    return True

remove_policies

remove_policies(sec: str, ptype: str, rules: list[list[str]]) -> None

Remove multiple policy rules from ShotGrid via batch operation.

PARAMETER	DESCRIPTION
sec	The policy section ("p" or "g"). TYPE: str
ptype	The policy type identifier. TYPE: str
rules	A list of policy rule value lists to remove. TYPE: list[list[str]]

Source code in shotgrid_casbin_adapter/core.py

def remove_policies(self, sec: str, ptype: str, rules: list[list[str]]) -> None:
    """Remove multiple policy rules from ShotGrid via batch operation.

    Args:
        sec: The policy section (``"p"`` or ``"g"``).
        ptype: The policy type identifier.
        rules: A list of policy rule value lists to remove.
    """
    if not rules:
        return
    all_ids: set[int] = set()
    for rule in rules:
        for e in self._sg.find(self._entity_type, _build_rule_filters(ptype, rule, self._project_id), ["id"]):
            all_ids.add(e["id"])  # type: ignore[typeddict-item]
    if all_ids:
        _batch_delete(self._sg, self._entity_type, [{"id": eid} for eid in all_ids])

remove_policy

remove_policy(sec: str, ptype: str, rule: list[str]) -> bool

Remove a policy rule from ShotGrid.

ShotGrid's delete retires the entity (soft-delete). ``find() excludes retired records by default.

PARAMETER	DESCRIPTION
sec	The policy section ("p" or "g"). TYPE: str
ptype	The policy type identifier. TYPE: str
rule	The policy rule values to remove. TYPE: list[str]

RETURNS	DESCRIPTION
bool	True if at least one matching rule was removed, False otherwise.

Source code in shotgrid_casbin_adapter/core.py

def remove_policy(self, sec: str, ptype: str, rule: list[str]) -> bool:
    """Remove a policy rule from ShotGrid.

    ShotGrid's delete retires the entity (soft-delete). ``find()
    excludes retired records by default.

    Args:
        sec: The policy section (``"p"`` or ``"g"``).
        ptype: The policy type identifier.
        rule: The policy rule values to remove.

    Returns:
        ``True`` if at least one matching rule was removed, ``False`` otherwise.
    """
    entities = self._sg.find(self._entity_type, _build_rule_filters(ptype, rule, self._project_id), ["id"])
    if not entities:
        return False
    for entity in entities:
        self._sg.delete(self._entity_type, entity["id"])  # type: ignore[typeddict-item]
    return True

save_policy

save_policy(model: Model) -> bool

Save all policy rules from the Casbin model to ShotGrid.

Replaces all existing rules with the current model state. First deletes all existing entities (optionally scoped to the configured project), then creates new ones for every rule in the model's "p" and "g" sections.

PARAMETER	DESCRIPTION
model	The Casbin model whose rules to persist. TYPE: Model

RETURNS	DESCRIPTION
bool	True on success.

Source code in shotgrid_casbin_adapter/core.py

def save_policy(self, model: Model) -> bool:
    """Save all policy rules from the Casbin model to ShotGrid.

    Replaces all existing rules with the current model state. First
    deletes all existing entities (optionally scoped to the configured
    project), then creates new ones for every rule in the model's
    ``"p"`` and ``"g"`` sections.

    Args:
        model: The Casbin model whose rules to persist.

    Returns:
        ``True`` on success.
    """
    existing = self._sg.find(self._entity_type, _project_filter(self._project_id), ["id"])
    _batch_delete(self._sg, self._entity_type, existing)

    create_requests: list[dict[str, Any]] = []
    for sec in ["p", "g"]:
        if sec not in model.model:
            continue
        for ptype, ast in model.model[sec].items():
            for rule in ast.policy:
                create_requests.append(_build_create_request(self._entity_type, ptype, rule, self._project_id))
    if create_requests:
        self._sg.batch(create_requests)
    return True

update_filtered_policies

update_filtered_policies(sec: str, ptype: str, new_rules: list[list[str]], field_index: int, *field_values: str) -> list[list[str]]

Update all policies matching a filter with new rules.

Deletes all entities matching the filter, then creates new entities for each rule in new_rules.

PARAMETER	DESCRIPTION
sec	The policy section ("p" or "g"). TYPE: str
ptype	The policy type identifier. TYPE: str
new_rules	The replacement policy rule value lists. TYPE: list[list[str]]
field_index	The starting field index (0-5) for filtering. TYPE: int
*field_values	The field values to match starting at field_index. TYPE: str DEFAULT: ()

RETURNS	DESCRIPTION
list[list[str]]	A list of the old rule value lists that were replaced.

Source code in shotgrid_casbin_adapter/core.py

def update_filtered_policies(
    self,
    sec: str,
    ptype: str,
    new_rules: list[list[str]],
    field_index: int,
    *field_values: str,
) -> list[list[str]]:
    """Update all policies matching a filter with new rules.

    Deletes all entities matching the filter, then creates new entities
    for each rule in ``new_rules``.

    Args:
        sec: The policy section (``"p"`` or ``"g"``).
        ptype: The policy type identifier.
        new_rules: The replacement policy rule value lists.
        field_index: The starting field index (0-5) for filtering.
        *field_values: The field values to match starting at ``field_index``.

    Returns:
        A list of the old rule value lists that were replaced.
    """
    filter_obj = Filter()
    filter_obj.ptype = [ptype]
    for i in range(len(field_values)):
        if field_index <= i < field_index + len(field_values):
            setattr(filter_obj, f"v{i}", [field_values[i - field_index]])
        else:
            break
    return self._update_filtered_policies(ptype, new_rules, filter_obj)

update_policies

update_policies(sec: str, ptype: str, old_rules: list[list[str]], new_rules: list[list[str]]) -> None

Update multiple policy rules in ShotGrid.

PARAMETER	DESCRIPTION
sec	The policy section ("p" or "g"). TYPE: str
ptype	The policy type identifier. TYPE: str
old_rules	The current policy rule value lists. TYPE: list[list[str]]
new_rules	The replacement policy rule value lists. TYPE: list[list[str]]

Source code in shotgrid_casbin_adapter/core.py

def update_policies(self, sec: str, ptype: str, old_rules: list[list[str]], new_rules: list[list[str]]) -> None:
    """Update multiple policy rules in ShotGrid.

    Args:
        sec: The policy section (``"p"`` or ``"g"``).
        ptype: The policy type identifier.
        old_rules: The current policy rule value lists.
        new_rules: The replacement policy rule value lists.
    """
    for i in range(len(old_rules)):
        self.update_policy(sec, ptype, old_rules[i], new_rules[i])

update_policy

update_policy(sec: str, ptype: str, old_rule: list[str], new_policy: list[str]) -> None

Update a policy rule in ShotGrid.

Finds the first entity matching old_rule and updates its fields to new_policy. If the new policy is shorter than the old rule, excess fields are set to None.

PARAMETER	DESCRIPTION
sec	The policy section ("p" or "g"). TYPE: str
ptype	The policy type identifier. TYPE: str
old_rule	The current policy rule values. TYPE: list[str]
new_policy	The replacement policy rule values. TYPE: list[str]

Source code in shotgrid_casbin_adapter/core.py

def update_policy(self, sec: str, ptype: str, old_rule: list[str], new_policy: list[str]) -> None:
    """Update a policy rule in ShotGrid.

    Finds the first entity matching ``old_rule`` and updates its fields
    to ``new_policy``. If the new policy is shorter than the old rule,
    excess fields are set to ``None``.

    Args:
        sec: The policy section (``"p"`` or ``"g"``).
        ptype: The policy type identifier.
        old_rule: The current policy rule values.
        new_policy: The replacement policy rule values.
    """
    entities = self._sg.find(self._entity_type, _build_rule_filters(ptype, old_rule, self._project_id), ["id"])
    if not entities:
        return

    data = _rule_to_dict(ptype, new_policy)
    for i in range(len(new_policy), max(len(old_rule), len(new_policy))):
        data[CASBIN_FIELDS[i + 1]] = None
    self._sg.update(self._entity_type, entities[0]["id"], data)  # type: ignore[typeddict-item]

Filter

Filter()

Filter for loading filtered policies from ShotGrid.

Each attribute is a list of values to match against the corresponding Casbin field. An empty list means the field is not filtered.

ATTRIBUTE	DESCRIPTION
ptype	List of policy type values to filter by. TYPE: list[str]
v0	List of v0 field values to filter by (maps to code). TYPE: list[str]
v1	List of v1 field values to filter by. TYPE: list[str]
v2	List of v2 field values to filter by. TYPE: list[str]
v3	List of v3 field values to filter by. TYPE: list[str]
v4	List of v4 field values to filter by. TYPE: list[str]
v5	List of v5 field values to filter by. TYPE: list[str]

Initialize a Filter with empty filter lists.

Source code in shotgrid_casbin_adapter/filter.py

def __init__(self) -> None:
    """Initialize a Filter with empty filter lists."""
    self.ptype: list[str] = []
    self.v0: list[str] = []
    self.v1: list[str] = []
    self.v2: list[str] = []
    self.v3: list[str] = []
    self.v4: list[str] = []
    self.v5: list[str] = []