swh.storage.proxies.masking.db module#

exception swh.storage.proxies.masking.db.DuplicateRequest[source]#: Bases: StorageArgumentException

exception swh.storage.proxies.masking.db.RequestNotFound[source]#: Bases: StorageArgumentException

class swh.storage.proxies.masking.db.MaskedState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

Value recording “how much” an object associated to a masking request is masked

VISIBLE = 1#: The object is visible

DECISION_PENDING = 2#: The object is temporarily masked until the request is reviewed

RESTRICTED = 3#: The access to the object is restricted permanently

class swh.storage.proxies.masking.db.MaskedStatus(state: MaskedState, request: UUID)[source]#

Bases: object

Return value when requesting if an object is masked

Method generated by attrs for class MaskedStatus.

class swh.storage.proxies.masking.db.MaskingRequest(id: UUID, slug: str, date: datetime, reason: str)[source]#

Bases: object

A request for masking a set of objects

Method generated by attrs for class MaskingRequest.

id#: Unique id for the request (will be returned to requesting clients)

slug#: Unique, human-readable id for the request (for administrative interactions)

date#: Date the request was received

reason#: Why the request was made

class swh.storage.proxies.masking.db.MaskingRequestHistory(request: UUID, date: datetime, message: str)[source]#

Bases: object

Method generated by attrs for class MaskingRequestHistory.

request#: id of the masking request

date#: Date the history entry has been added

message#: Free-form history information (e.g. “policy decision made”)

class swh.storage.proxies.masking.db.MaskedObject(request_slug: str, swhid: ExtendedSWHID, state: MaskedState)[source]#

Bases: object

Method generated by attrs for class MaskedObject.

class swh.storage.proxies.masking.db.MaskingDb(*args, **kwargs)[source]#

Bases: BaseDb

create a DB proxy

Parameters:

conn – psycopg2 connection to the SWH DB
pool – psycopg2 pool of connections

class swh.storage.proxies.masking.db.MaskingAdmin(*args, **kwargs)[source]#

Bases: MaskingDb

create a DB proxy

Parameters:

conn – psycopg2 connection to the SWH DB
pool – psycopg2 pool of connections

create_request(slug: str, reason: str) → MaskingRequest[source]#

Record a new masking request

Parameters:

slug – human-readable unique identifier for the request
reason – free-form text recording why the request was made

Raises:

DuplicateRequest when the slug already exists –

find_request(slug: str) → MaskingRequest | None[source]#

Find a masking request using its slug

Returns: None if a request with the given slug doesn’t exist

find_request_by_id(id: UUID) → MaskingRequest | None[source]#

Find a masking request using its id

Returns: None if a request with the given request doesn’t exist

get_requests(include_cleared_requests: bool = False) → List[Tuple[MaskingRequest, int]][source]#

Get known requests

Parameters:

include_cleared_requests – also include requests with no associated
states (masking)

set_object_state(request_id: UUID, new_state: MaskedState, swhids: List[ExtendedSWHID])[source]#

Within the request with the given id, record the state of the given objects as new_state.

This creates entries or updates them as appropriate.

Raises: RequestNotFound if the request is not found.

get_states_for_request(request_id: UUID) → Dict[ExtendedSWHID, MaskedState][source]#

Get the state of objects associated with the given request.

Raises RequestNotFound if the request is not found.

find_masks(swhids: List[ExtendedSWHID]) → List[MaskedObject][source]#: Lookup the masking state and associated requests for the given SWHIDs.

delete_masks(request_id: UUID) → None[source]#

Remove all masking states for the given request.

Raises: RequestNotFound if the request is not found.

record_history(request_id: UUID, message: str) → MaskingRequestHistory[source]#

Add an entry to the history of the given request.

Raises: RequestNotFound if the request is not found.

get_history(request_id: UUID) → List[MaskingRequestHistory][source]#

Get the history of a given request.

Raises: RequestNotFound if the request if not found.

class swh.storage.proxies.masking.db.MaskingQuery(*args, **kwargs)[source]#

Bases: MaskingDb

create a DB proxy

Parameters:

conn – psycopg2 connection to the SWH DB
pool – psycopg2 pool of connections

swhids_are_masked(swhids: List[ExtendedSWHID]) → Dict[ExtendedSWHID, List[MaskedStatus]][source]#

Checks which objects in the list are masked.

Returns: For each masked object, a list of MaskedStatus objects where the State is not MaskedState.VISIBLE.