Skip to content

oarepo_oaipmh_harvester.permissions

oarepo_oaipmh_harvester.permissions

Permission policies and generators for harvesters and harvest records.

Both manager generators also provide search query filters, so that managers only find their own harvesters / records.

HarvestManager

Bases: Generator

Grants harvest managers access to the harvesters they manage.

Source code in oarepo_oaipmh_harvester/permissions.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class HarvestManager(Generator):
    """Grants harvest managers access to the harvesters they manage."""

    @override
    def needs(self, **kwargs: Any) -> Collection[Need]:
        """Return the needs for harvest managers."""
        if "record" not in kwargs:
            return []
        record = kwargs["record"]
        # the record is an aggregate (its dict data is empty), so read the model field
        return [harvest_manager_need(manager) for manager in (record.harvest_managers or []) if "user" in manager]

    @override
    def query_filter(self, **kwargs: Any) -> dsl.query.Query | list[dsl.query.Query] | None:
        """Restrict search to harvesters managed by the identity."""
        identity = kwargs["identity"]
        if not identity or not identity.id:
            return MatchNone()
        return Term(**{"harvest_managers.user": identity.id})

needs(**kwargs: Any) -> Collection[Need]

Return the needs for harvest managers.

Source code in oarepo_oaipmh_harvester/permissions.py 
52
53
54
55
56
57
58
59
@override
def needs(self, **kwargs: Any) -> Collection[Need]:
    """Return the needs for harvest managers."""
    if "record" not in kwargs:
        return []
    record = kwargs["record"]
    # the record is an aggregate (its dict data is empty), so read the model field
    return [harvest_manager_need(manager) for manager in (record.harvest_managers or []) if "user" in manager]

query_filter(**kwargs: Any) -> dsl.query.Query | list[dsl.query.Query] | None

Restrict search to harvesters managed by the identity.

Source code in oarepo_oaipmh_harvester/permissions.py 
61
62
63
64
65
66
67
@override
def query_filter(self, **kwargs: Any) -> dsl.query.Query | list[dsl.query.Query] | None:
    """Restrict search to harvesters managed by the identity."""
    identity = kwargs["identity"]
    if not identity or not identity.id:
        return MatchNone()
    return Term(**{"harvest_managers.user": identity.id})

OAIHarvesterPermissionPolicy

Bases: RecordPermissionPolicy

Permission policy for OAI Harvester records.

Source code in oarepo_oaipmh_harvester/permissions.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class OAIHarvesterPermissionPolicy(RecordPermissionPolicy):
    """Permission policy for OAI Harvester records."""

    can_search = (SystemProcess(), AnyUser())
    can_read = (SystemProcess(), AdministrationWithQueryFilter(), HarvestManager())
    can_create = (SystemProcess(), AdministrationWithQueryFilter())
    can_update = (SystemProcess(), AdministrationWithQueryFilter())
    can_delete = (SystemProcess(), AdministrationWithQueryFilter())
    can_manage = (SystemProcess(), AdministrationWithQueryFilter())
    can_run_harvest = (
        SystemProcess(),
        AdministrationWithQueryFilter(),
        HarvestManager(),
    )

    can_create_files = (SystemProcess(), AdministrationWithQueryFilter())
    can_set_content_files = (SystemProcess(), AdministrationWithQueryFilter())
    can_get_content_files = (SystemProcess(), AdministrationWithQueryFilter())
    can_commit_files = (SystemProcess(), AdministrationWithQueryFilter())
    can_read_files = (SystemProcess(), AdministrationWithQueryFilter())
    can_update_files = (SystemProcess(), AdministrationWithQueryFilter())
    can_delete_files = (SystemProcess(), AdministrationWithQueryFilter())

    can_edit = (SystemProcess(),)
    can_new_version = (SystemProcess(),)
    can_search_drafts = (SystemProcess(),)
    can_read_draft = (SystemProcess(),)
    can_update_draft = (SystemProcess(),)
    can_delete_draft = (SystemProcess(),)
    can_publish = (SystemProcess(),)
    can_draft_create_files = (SystemProcess(),)
    can_draft_set_content_files = (SystemProcess(),)
    can_draft_get_content_files = (SystemProcess(),)
    can_draft_commit_files = (SystemProcess(),)
    can_draft_read_files = (SystemProcess(),)
    can_draft_update_files = (SystemProcess(),)

HarvestRecordManager

Bases: Generator

Generator for giving access to harvested records for harvest record managers.

Source code in oarepo_oaipmh_harvester/permissions.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class HarvestRecordManager(Generator):
    """Generator for giving access to harvested records for harvest record managers."""

    @override
    def needs(self, **kwargs: Any) -> Collection[Need]:
        """Return the needs for harvest record managers."""
        if "record" not in kwargs:
            return []

        harvester = kwargs["record"].harvester

        return [harvest_manager_need(manager) for manager in (harvester.harvest_managers or []) if "user" in manager]

    def query_filter(self, **kwargs: Any) -> dsl.query.Query | list[dsl.query.Query] | None:
        """Restrict search to records of harvesters managed by the identity."""
        identity = kwargs["identity"]
        if not identity or not identity.id:
            return MatchNone()
        return Term(**{"harvest_managers.user": identity.id})

needs(**kwargs: Any) -> Collection[Need]

Return the needs for harvest record managers.

Source code in oarepo_oaipmh_harvester/permissions.py 
111
112
113
114
115
116
117
118
119
@override
def needs(self, **kwargs: Any) -> Collection[Need]:
    """Return the needs for harvest record managers."""
    if "record" not in kwargs:
        return []

    harvester = kwargs["record"].harvester

    return [harvest_manager_need(manager) for manager in (harvester.harvest_managers or []) if "user" in manager]

query_filter(**kwargs: Any) -> dsl.query.Query | list[dsl.query.Query] | None

Restrict search to records of harvesters managed by the identity.

Source code in oarepo_oaipmh_harvester/permissions.py 
121
122
123
124
125
126
def query_filter(self, **kwargs: Any) -> dsl.query.Query | list[dsl.query.Query] | None:
    """Restrict search to records of harvesters managed by the identity."""
    identity = kwargs["identity"]
    if not identity or not identity.id:
        return MatchNone()
    return Term(**{"harvest_managers.user": identity.id})

OAIRecordPermissionPolicy

Bases: RecordPermissionPolicy

Permission policy for harvested records.

Source code in oarepo_oaipmh_harvester/permissions.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class OAIRecordPermissionPolicy(RecordPermissionPolicy):
    """Permission policy for harvested records."""

    can_create = (SystemProcess(),)
    can_read = (
        SystemProcess(),
        AdministrationWithQueryFilter(),
        HarvestRecordManager(),
    )
    can_run_harvest = (
        SystemProcess(),
        AdministrationWithQueryFilter(),
        HarvestRecordManager(),
    )
    can_search = (AuthenticatedUser(), SystemProcess())
    can_update = (SystemProcess(),)
    can_delete = (SystemProcess(),)

harvest_manager_need(manager: dict[str, Any]) -> Need

Return the need of a harvest manager entry ({"user": id}).

The user id is stored as a string, but the identity provides UserNeed with an integer user id, so it is converted here to make the needs match.

Source code in oarepo_oaipmh_harvester/permissions.py 
40
41
42
43
44
45
46
def harvest_manager_need(manager: dict[str, Any]) -> Need:
    """Return the need of a harvest manager entry (`{"user": id}`).

    The user id is stored as a string, but the identity provides `UserNeed` with
    an integer user id, so it is converted here to make the needs match.
    """
    return UserNeed(int(manager["user"]))