generated from Lookyloo/pyproject_template
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
new: make sure to always catch exceptions in tasks
- Loading branch information
Showing
6 changed files
with
283 additions
and
218 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| #!/usr/bin/env python3 | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from enum import IntEnum, unique | ||
| from logging import LoggerAdapter | ||
| from typing import MutableMapping, Any, TypedDict | ||
|
|
||
| from playwrightcapture.capture import CaptureResponse as PlaywrightCaptureResponse | ||
|
|
||
|
|
||
| class LacusCoreException(Exception): | ||
| pass | ||
|
|
||
|
|
||
| class CaptureError(LacusCoreException): | ||
| pass | ||
|
|
||
|
|
||
| class RetryCapture(LacusCoreException): | ||
| pass | ||
|
|
||
|
|
||
| class CaptureSettingsError(LacusCoreException): | ||
| pass | ||
|
|
||
|
|
||
| class LacusCoreLogAdapter(LoggerAdapter): # type: ignore[type-arg] | ||
| """ | ||
| Prepend log entry with the UUID of the capture | ||
| """ | ||
| def process(self, msg: str, kwargs: MutableMapping[str, Any]) -> tuple[str, MutableMapping[str, Any]]: | ||
| if self.extra: | ||
| return '[{}] {}'.format(self.extra['uuid'], msg), kwargs | ||
| return msg, kwargs | ||
|
|
||
|
|
||
| @unique | ||
| class CaptureStatus(IntEnum): | ||
| '''The status of the capture''' | ||
| UNKNOWN = -1 | ||
| QUEUED = 0 | ||
| DONE = 1 | ||
| ONGOING = 2 | ||
|
|
||
|
|
||
| class CaptureResponse(PlaywrightCaptureResponse, TypedDict, total=False): | ||
| '''A capture made by Lacus. With the base64 encoded image and downloaded file decoded to bytes.''' | ||
|
|
||
| # Need to make sure the type is what's expected down the line | ||
| children: list[CaptureResponse] | None # type: ignore[misc] | ||
|
|
||
| status: int | ||
| runtime: float | None | ||
|
|
||
|
|
||
| class CaptureResponseJson(TypedDict, total=False): | ||
| '''A capture made by Lacus. With the base64 encoded image and downloaded file *not* decoded.''' | ||
|
|
||
| status: int | ||
| last_redirected_url: str | None | ||
| har: dict[str, Any] | None | ||
| cookies: list[dict[str, str]] | None | ||
| error: str | None | ||
| html: str | None | ||
| png: str | None | ||
| downloaded_filename: str | None | ||
| downloaded_file: str | None | ||
| children: list[CaptureResponseJson] | None | ||
| runtime: float | None | ||
| potential_favicons: list[str] | None | ||
|
|
||
|
|
||
| class CaptureSettings(TypedDict, total=False): | ||
| '''The capture settings that can be passed to Lacus.''' | ||
|
|
||
| url: str | None | ||
| document_name: str | None | ||
| document: str | None | ||
| browser: str | None | ||
| device_name: str | None | ||
| user_agent: str | None | ||
| proxy: str | dict[str, str] | None | ||
| general_timeout_in_sec: int | None | ||
| cookies: list[dict[str, Any]] | None | ||
| headers: str | dict[str, str] | None | ||
| http_credentials: dict[str, str] | None | ||
| geolocation: dict[str, float] | None | ||
| timezone_id: str | None | ||
| locale: str | None | ||
| color_scheme: str | None | ||
| viewport: dict[str, int] | None | ||
| referer: str | None | ||
| with_favicon: bool | ||
| allow_tracking: bool | ||
| force: bool | ||
| recapture_interval: int | ||
| priority: int | ||
| uuid: str | None | ||
|
|
||
| depth: int | ||
| rendered_hostname_only: bool # Note: only used if depth is > 0 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| #!/usr/bin/env python3 | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from typing import Any, Coroutine, Optional, TypeVar, Tuple | ||
|
|
||
| import asyncio | ||
| import functools | ||
| import logging | ||
|
|
||
| from .helpers import LacusCoreLogAdapter | ||
|
|
||
| T = TypeVar('T') | ||
|
|
||
| # Code from https://quantlane.com/blog/ensure-asyncio-task-exceptions-get-logged/ | ||
|
|
||
|
|
||
| def create_task( | ||
| coroutine: Coroutine[Any, Any, T], | ||
| *, | ||
| name: str, | ||
| logger: 'LacusCoreLogAdapter', | ||
| message: str, | ||
| message_args: Tuple[Any, ...] = (), | ||
| loop: Optional[asyncio.AbstractEventLoop] = None, | ||
|
|
||
| ) -> 'asyncio.Task[T]': # This type annotation has to be quoted for Python < 3.9, see https://www.python.org/dev/peps/pep-0585/ | ||
| ''' | ||
| This helper function wraps a ``loop.create_task(coroutine())`` call and ensures there is | ||
| an exception handler added to the resulting task. If the task raises an exception it is logged | ||
| using the provided ``logger``, with additional context provided by ``message`` and optionally | ||
| ``message_args``. | ||
| ''' | ||
| if loop is None: | ||
| loop = asyncio.get_running_loop() | ||
| task = loop.create_task(coroutine, name=name) | ||
| task.add_done_callback( | ||
| functools.partial(_handle_task_result, logger=logger, message=message, message_args=message_args) | ||
| ) | ||
| return task | ||
|
|
||
|
|
||
| def _handle_task_result( | ||
| task: asyncio.Task[Any], | ||
| *, | ||
| logger: logging.Logger, | ||
| message: str, | ||
| message_args: Tuple[Any, ...] = (), | ||
| ) -> None: | ||
| try: | ||
| task.result() | ||
| except asyncio.CancelledError: | ||
| pass # Task cancellation should not be logged as an error. | ||
| except asyncio.TimeoutError: | ||
| pass # Timeout is also fine | ||
| # Ad the pylint ignore: we want to handle all exceptions here so that the result of the task | ||
| # is properly logged. There is no point re-raising the exception in this callback. | ||
| except Exception: # pylint: disable=broad-except | ||
| logger.exception(message, *message_args) |
Oops, something went wrong.