Quick look at the repo for arq by Samuel Colvin.

Codemap

  • .github/: mainly GitHub Actions
  • arq/: source code
  • docs/: sphinx documentation
  • requirements/: requirements.in and txt, using pip-tools, for local dev
  • tests/: pytest, one test file per public module
  • .codecov.yml: codecov config
  • .gitignore
  • .pre-commit-config.yaml: pre-commit + make lint + mypy
  • HISTORY.rst: changelog
  • LICENSE: MIT
  • Makefile: install, format, lint (flake8, isort, black), test, all, docs, etc
  • README.md: simple readme for github and pypi
  • pyproject.toml: hatchling, PEP 621, config for pytest, coverage, black, isort, mypy

Dependencies

Direct

  • redis[hiredis]: redis interface with faster parser
  • click: Command Line Interface Creation Kit, from the pallets team
  • typing-extensions: backport of new typing features for older python versions

docs.in

  • Sphinx: documentation generator for multiple languages

linting.in

  • black: code formatter
  • flake8: linter
  • flake8-quotes
  • isort[colors]: import sorter with colorama
  • mypy: static type checker
  • types-pytz: typeshed stubs for pytz
  • types-redis: typeshed stubs for redis

testing.in

  • coverage[toml]
  • dirty-equals
  • msgpack
  • pydantic
  • pytest
  • pytest-asyncio
  • pytest-mock
  • pytest-sugar
  • pytest-timeout
  • pytz
  • redislite

Code examples

  • Use __all__ in __init__.py to control what is exported
__all__ = (
    'ArqRedis',
    'create_pool',
    'cron',
    'VERSION',
    'Retry',
    'Worker',
    'check_health',
    'func',
    'run_worker',
)
  • __main__.py file to run the module as a script, e.g. arq --help:
from .cli import cli

if __name__ == '__main__':
    cli()
  • from typing import TYPE_CHECKING to avoid circular imports:
if TYPE_CHECKING:
    from .typing import WorkerSettingsType
  • py.typed file to indicate that the package is fully typed, for tools like mypy
  • sphinx markup in docstrings, type only with type hints
async def abort(self, *, timeout: Optional[float] = None, poll_delay: float = 0.5) -> bool:
    """
    Abort the job.

    :param timeout: maximum time to wait for the job result before raising ``TimeoutError``,
        will wait forever on None
    :param poll_delay: how often to poll redis for the job result
    :return: True if the job aborted properly, False otherwise
    """
  • Use sys.version_info to check for python version
if sys.version_info >= (3, 8):
    from typing import Literal, Protocol
else:
    from typing_extensions import Literal, Protocol
  • typing.py file with type aliases and Protocol classes instead of Callable
OptionType = Union[None, Set[int], int]
WeekdayOptionType = Union[OptionType, Literal['mon', 'tues', 'wed', 'thurs', 'fri', 'sat', 'sun']]
SecondsTimedelta = Union[int, float, timedelta]

class WorkerCoroutine(Protocol):
    __qualname__: str

    async def __call__(self, ctx: Dict[Any, Any], *args: Any, **kwargs: Any) -> Any:  # pragma: no cover
        pass
  • Use cast to avoid typing errors
redis_settings = cast(Optional[RedisSettings], cls_kwargs.get('redis_settings'))
health_check_key = cast(Optional[str], cls_kwargs.get('health_check_key'))
queue_name = cast(Optional[str], cls_kwargs.get('queue_name'))
  • Keyword only arguments using *
async def create_pool(
    settings_: RedisSettings = None,
    *,
    retry: int = 0,
    job_serializer: Optional[Serializer] = None,
    job_deserializer: Optional[Deserializer] = None,
    default_queue_name: str = default_queue_name,
    expires_extra_ms: int = expires_extra_ms,
) -> ArqRedis: