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.DisplayName(original_email: bytes, display_name: bytes)[source]#

Bases: object

A request for masking a set of objects

Method generated by attrs for class DisplayName.

original_email#

Email on revision/release objects to match before applying the display name

display_name#

Full name, usually of the form Name <email>, used for display queries

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

current_version = 194#
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.

set_display_name(original_email: bytes, display_name: bytes) → None[source]#

Updates the display name of the person identified by the email address.

set_display_names(display_names: Iterable[Tuple[bytes, bytes]], clear=False) → None[source]#

Insert a list of display names of persons identified by their email address.

If ‘clear’ is True, empty the existing display names before inserting the new entries.

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.

display_name(original_emails: List[bytes]) → Dict[bytes, bytes][source]#

Returns the display name of the person identified by each original_email, if any.

iter_masked_swhids() → Iterator[Tuple[ExtendedSWHID, List[MaskedStatus]]][source]#

Returns the complete list of masked SWHIDs.

SWHIDs are guaranteed to be unique in the iterator.

Yields:

For each masked object, its SWHID and a list of MaskedStatus objects where the State is not MaskedState.VISIBLE.