core.utils.fetchers.SafePOPMixin module

Module with the SafePOPMixin mixin.

type core.utils.fetchers.SafePOPMixin.POP3Response = bytes | tuple[bytes, list[bytes], int]

class core.utils.fetchers.SafePOPMixin.POP3FetcherClass(*args, **kwargs)

Bases: Protocol

Protocol defining the required attributes of a class implementing this mixin.

_mail_client: poplib.POP3

logger: logging.Logger

__init__(*args, **kwargs)

_is_protocol = True

class core.utils.fetchers.SafePOPMixin.SafePOPMixin

Bases: object

Mixin with safe POP operations.

The implementing class must have an _mail_client that is an poplib.POP3 or poplib.POP3_SSL and an instance logger.

Only errors leading up to and during the fetching process should raise outside of the class. Otherwise issues with logout etc would destroy the work done with fetching, which makes no sense.

check_response(response, command_name, exception_class, expected_status=b'+OK')

Checks the status response of a POP action.

If it doesn’t match the expectation raises an exception.

Todo

Safely! extract error message from response for logging and exception.

Parameters:

• response (bytes | tuple[bytes, list[bytes], int]) – The complete response to the POP action.

• command_name (str) – The name of the action.

• exception_class (type[FetcherError] | None) – The exception class to raise if the status doesn’t match the expectation. Defaults to core.utils.fetchers.exceptions.FetcherError.

• expected_status (bytes) – The expected status response. Defaults to “+OK”.

Raises:

exception_class – If the response status doesn’t match the expectation.

Return type:

None

static safe(exception_class, expected_status=b'+OK')

Overloads:

• exception_class (type[FetcherError]), expected_status (bytes) → Callable[[Callable[…, POP3ActionResponse]], Callable[…, POP3ActionResponse]]

• exception_class (None), expected_status (bytes) → Callable[[Callable[…, POP3ActionResponse]], Callable[…, POP3ActionResponse | None]]

Parameters:

• exception_class (type[FetcherError] | None)

• expected_status (bytes)

Return type:

Callable[[Callable[[…], POP3ActionResponse]], Callable[[…], POP3ActionResponse | None]]

Wrapper for POP actions.

Catches expected errors and checks for correct responses and raises an FetcherError.

Todo

Find a better way to disable the exception and getting rid of the typing mess.

Parameters:

• exception_class (type[FetcherError] | None) – The exception class to raise if an error occurs or the status doesn’t match the expectation. Must be a subclass of core.utils.fetchers.exceptions.FetcherError. Defaults to core.utils.fetchers.exceptions.FetcherError. If None is passed, no exception is raised, all errors are logged nonetheless.

• expected_status (bytes) – The expected status response. Defaults to “+OK”.

Returns:

The return value of the wrapped action. None if an error occurs and exception is None.

Raises:

exception_class – If an error occurs or the status doesn’t match the expectation.

Return type:

Callable[[Callable[[…], POP3ActionResponse]], Callable[[…], POP3ActionResponse | None]]

safe_user(*args, **kwargs)

Return type:

TypeVar(POP3ActionResponse, bound= bytes | tuple[bytes, list[bytes], int]) | None

Parameters:

• self (Self)

• args (Any)

• kwargs (Any)

safe_pass_(*args, **kwargs)

Return type:

TypeVar(POP3ActionResponse, bound= bytes | tuple[bytes, list[bytes], int]) | None

Parameters:

• self (Self)

• args (Any)

• kwargs (Any)

safe_noop(*args, **kwargs)

Return type:

TypeVar(POP3ActionResponse, bound= bytes | tuple[bytes, list[bytes], int]) | None

Parameters:

• self (Self)

• args (Any)

• kwargs (Any)

safe_list(*args, **kwargs)

Return type:

TypeVar(POP3ActionResponse, bound= bytes | tuple[bytes, list[bytes], int]) | None

Parameters:

• self (Self)

• args (Any)

• kwargs (Any)

safe_retr(*args, **kwargs)

Return type:

TypeVar(POP3ActionResponse, bound= bytes | tuple[bytes, list[bytes], int]) | None

Parameters:

• self (Self)

• args (Any)

• kwargs (Any)

safe_quit(*args, **kwargs)

Return type:

TypeVar(POP3ActionResponse, bound= bytes | tuple[bytes, list[bytes], int]) | None

Parameters:

• self (Self)

• args (Any)

• kwargs (Any)