feat: add async run_subprocess with timeout and flag features and more utilities (#24)

Author: ticdenis

.github/workflows/cd.yml

@@ -22,7 +22,7 @@ jobs:
       - name: version
         run: sed -i "s/__version__ = '.*'/__version__ = '$VERSION'/g" aioddd/__init__.py
       - name: deps
-        run: python3 run-script dev-install
+        run: python3 -m pip install .[dev,deploy]
       - name: deploy
         env:
           TWINE_USERNAME: ${{ secrets.POETRY_HTTP_BASIC_PYPI_USERNAME }}

.github/workflows/ci.yml

@@ -22,7 +22,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
       - name: deps
         timeout-minutes: 5
-        run: python3 run-script dev-install
+        run: python3 -m pip install .[dev,fmt,security-analysis,static-analysis,test]
       - name: security-analysis
         timeout-minutes: 1
         run: python3 run-script security-analysis

Dockerfile

@@ -1,25 +1,24 @@
-FROM docker.io/library/python:3.7-slim AS production
+FROM docker.io/library/python:3.7-slim AS base
 
-WORKDIR /app
+RUN apt update -y && python3 -m pip install --upgrade pip
 
-COPY LICENSE README.md pyproject.toml ./
+WORKDIR /app
 
-RUN apt update -y &&  \
-    python3 -m pip install --upgrade pip && \
-    python3 -m pip install .
+FROM base AS production
 
+COPY LICENSE README.md pyproject.toml ./
 COPY aioddd ./aioddd/
 
+RUN python3 -m pip install .
+
 ENTRYPOINT ["python3"]
 CMD []
 
 FROM production AS development
 
-RUN apt install -y gcc
-
 COPY .pre-commit-config.yaml run-script ./
 
-RUN python3 run-script dev-install
+RUN python3 -m pip install .[dev,deploy,docs,fmt,security-analysis,static-analysis,test]
 
 COPY docs_src ./docs_src
 COPY tests ./tests

aioddd/__init__.py

@@ -43,7 +43,17 @@
     find_event_mapper_by_name,
     find_event_mapper_by_type,
 )
-from .utils import env, get_env, get_simple_logger
+from .subprocess import SubprocessResult, run_subprocess  # nosec
+from .utils import (
+    env,
+    get_bool_env,
+    get_env,
+    get_float_env,
+    get_int_env,
+    get_list_str_env,
+    get_simple_logger,
+    get_str_env,
+)
 from .value_objects import Id, StrDateTime, Timestamp
 
 __version__ = '1.3.7'
@@ -92,10 +102,18 @@
     'InternalEventPublisher',
     # utils
     'get_env',
+    'get_str_env',
+    'get_bool_env',
+    'get_int_env',
+    'get_float_env',
+    'get_list_str_env',
     'get_simple_logger',
     'env',
     # value_objects
     'Id',
     'Timestamp',
     'StrDateTime',
+    # subprocess,
+    'SubprocessResult',
+    'run_subprocess',
 )

aioddd/subprocess/__init__.py

@@ -0,0 +1,94 @@
+from asyncio import TimeoutError as AsyncTimeoutError
+from asyncio import create_subprocess_exec, create_subprocess_shell, sleep, wait_for
+from typing import Any, Dict, NamedTuple, Optional
+
+_flag_subprocess_running: Dict[str, bool] = {}
+
+DEFAULT_WAIT_FLAG_SLEEP: float = 0.1
+
+
+async def _wait_flagged_subprocess_running(wait_flag: str) -> None:
+    if not _flag_subprocess_running[wait_flag]:
+        return
+    await sleep(DEFAULT_WAIT_FLAG_SLEEP)
+
+
+class SubprocessResult(NamedTuple):
+    return_code: int
+    stdout: Optional[str] = None
+    stderr: Optional[str] = None
+
+
+def _create_subprocess(  # type: ignore
+    *args: Any,
+    shell: bool,
+    stdout: Optional[int],
+    stderr: Optional[int],
+    limit: int,
+    **kwds: Any,
+):
+    return (
+        create_subprocess_shell(*args, stdout=stdout, stderr=stderr, limit=limit, **kwds)  # type: ignore
+        if shell
+        else create_subprocess_exec(*args, stdout=stdout, stderr=stderr, limit=limit, **kwds)
+    )
+
+
+async def run_subprocess(
+    *args: str,
+    shell: bool = False,
+    encoding: str = 'utf8',
+    timeout: Optional[float] = None,
+    wait_flag: Optional[str] = None,
+    wait_flag_timeout: Optional[float] = None,
+    stdout: Optional[int] = -1,  # see asyncio.subprocess.PIPE,
+    stderr: Optional[int] = -1,  # see asyncio.subprocess.PIPE,
+    limit: int = 2**64,  # see streams._DEFAULT_LIMIT
+    **kwds: Dict[str, Any],
+) -> SubprocessResult:
+    """
+    Creates and runs a subprocess with or without shell.
+
+    Provides to time out the Python managed subprocess using asyncio.wait_for.
+    Provides to flag Python managed subprocess with timeout as well using unique keys and asyncio.wait_for.
+
+    stdin not supported!
+
+    When shell=True *args will be the "cmd" arg for asyncio.subprocess.create_subprocess_shell
+    When shell=False *args will be the "*args" arg (not "program" arg) for asyncio.subprocess.create_subprocess_exec
+
+    Return (return_code, stdout, stderr)
+    """
+    _ = [kwds.pop(key, None) for key in ['shell', 'encoding', 'timeout', 'wait_flag', 'wait_flag_timeout']]
+    if wait_flag:
+        if wait_flag not in _flag_subprocess_running:
+            _flag_subprocess_running[wait_flag] = True
+        elif _flag_subprocess_running.get(wait_flag, True):
+            await wait_for(fut=_wait_flagged_subprocess_running(wait_flag=wait_flag), timeout=wait_flag_timeout)
+            _flag_subprocess_running[wait_flag] = True
+    proc = await _create_subprocess(
+        *args,
+        shell=shell,  # nosec
+        stdout=stdout,
+        stderr=stderr,
+        limit=limit,
+        **kwds,
+    )
+
+    try:
+        return_code = await wait_for(fut=proc.wait(), timeout=timeout)
+    except AsyncTimeoutError:
+        proc.terminate()
+        return_code = await proc.wait()
+    finally:
+        if wait_flag and wait_flag in _flag_subprocess_running:
+            del _flag_subprocess_running[wait_flag]
+
+    stdout_: Optional[bytes] = await proc.stdout.read()
+    stderr_: Optional[bytes] = await proc.stderr.read()
+
+    return SubprocessResult(
+        return_code=return_code,
+        stdout=stdout_.strip().decode(encoding) if stdout_ else None,
+        stderr=stderr_.strip().decode(encoding) if stderr_ else None,
+    )

aioddd/utils.py

@@ -1,6 +1,6 @@
 from logging import NOTSET, Formatter, Logger, StreamHandler, getLogger
 from os import getenv
-from typing import Any, Dict, Optional, Type, TypeVar, Union
+from typing import Any, Dict, List, Optional, Type, TypeVar, Union, cast
 
 
 def get_env(key: str, default: Optional[str] = None, cast_default_to_str: bool = True) -> Optional[str]:
@@ -11,6 +11,38 @@ def get_env(key: str, default: Optional[str] = None, cast_default_to_str: bool =
     return value
 
 
+def get_str_env(key: str, default: str = '') -> str:
+    return cast(str, get_env(key=key, default=default, cast_default_to_str=True))
+
+
+_boolean_positive_values: List[str] = ['True', 'true', 'yes', 'Y', 'y', '1']
+
+
+def get_bool_env(key: str, default: Union[bool, int] = False) -> bool:
+    return get_env(key=key, default=str(int(default)), cast_default_to_str=False) in _boolean_positive_values
+
+
+def get_int_env(key: str, default: Union[bool, int] = 0) -> int:
+    val = cast(str, get_env(key=key, default=str(int(default)), cast_default_to_str=False))
+    return int(val) if val.isdigit() else default
+
+
+def get_float_env(key: str, default: Union[bool, int, float] = 0) -> float:
+    val = cast(str, get_env(key=key, default=str(float(default)), cast_default_to_str=False))
+    return float(val) if val.isdigit() or val.replace('.', '').isdigit() else default
+
+
+def get_list_str_env(
+    key: str,
+    default: Optional[List[str]] = None,
+    *,
+    delimiter: str = ',',
+    allow_empty: bool = True,
+) -> List[str]:
+    val = cast(str, get_env(key=key, default='' if allow_empty else delimiter.join(default or [])))
+    return [] if allow_empty and (val is None or len(val) == 0) else val.split(delimiter)
+
+
 def get_simple_logger(
     name: Optional[str] = None,
     level: Union[str, int] = NOTSET,

docs_src/docs/en/index.md

@@ -12,7 +12,7 @@ Key Features:
 
 ## Requirements
 
-- Python 3.6+
+- Python 3.7+
 
 ## Installation
 
@@ -84,4 +84,4 @@ if __name__ == '__main__':
 [MIT](https://github.com/aiopy/python-aioddd/blob/master/LICENSE)
 
 
-### WIP
+### WIP

docs_src/docs/es/index.md

@@ -12,7 +12,7 @@
 
 ## Requisitos
 
-- Python 3.6+
+- Python 3.7+
 
 ## Instalación
 
@@ -84,4 +84,4 @@ if __name__ == '__main__':
 [MIT](https://github.com/aiopy/python-aioddd/blob/master/LICENSE)
 
 
-### WIP
+### WIP

docs_src/generated/en/index.html

@@ -433,7 +433,7 @@ <h1 id="async-python-ddd-utilities-library">Async Python DDD utilities library</
 </ul>
 <h2 id="requirements">Requirements</h2>
 <ul>
-<li>Python 3.6+</li>
+<li>Python 3.7+</li>
 </ul>
 <h2 id="installation">Installation</h2>
 <div class="highlight"><pre><span></span><code>python3 -m pip install aioddd

docs_src/generated/en/search/search_index.json

@@ -1 +1 @@
-{"config":{"indexing":"full","lang":["en"],"min_search_length":3,"prebuild_index":false,"separator":"[\\s\\-]+"},"docs":[{"location":"","text":"Async Python DDD utilities library Key Features: Aggregates : Aggregate & AggregateRoot ValueObjects : Id, Timestamp & StrDateTime CQRS : Command, CommandBus, SimpleCommandBus, Query, Response, QueryHandler, QueryBus & SimpleQueryBus EventSourcing : Event, EventMapper, EventPublisher, EventHandler, EventBus, SimpleEventBus & InternalEventPublisher Errors : raise_, BaseError, NotFoundError, ConflictError, BadRequestError, UnauthorizedError, ForbiddenError, UnknownError, IdInvalidError, TimestampInvalidError, DateTimeInvalidError, EventMapperNotFoundError, EventNotPublishedError, CommandNotRegisteredError & QueryNotRegisteredError Tests : AsyncMock & mock Utils : get_env & get_simple_logger Requirements Python 3.6+ Installation python3 -m pip install aioddd Example from asyncio import get_event_loop from dataclasses import dataclass from typing import Type from aioddd import NotFoundError , \\ Command , CommandHandler , SimpleCommandBus , \\ Query , QueryHandler , OptionalResponse , SimpleQueryBus , Event _products = [] class ProductStored ( Event ): @dataclass class Attributes : ref : str attributes : Attributes class StoreProductCommand ( Command ): def __init__ ( self , ref : str ): self . ref = ref class StoreProductCommandHandler ( CommandHandler ): def subscribed_to ( self ) -> Type [ Command ]: return StoreProductCommand async def handle ( self , command : StoreProductCommand ) -> None : _products . append ( command . ref ) class ProductNotFoundError ( NotFoundError ): _code = 'product_not_found' _title = 'Product not found' class FindProductQuery ( Query ): def __init__ ( self , ref : str ): self . ref = ref class FindProductQueryHandler ( QueryHandler ): def subscribed_to ( self ) -> Type [ Query ]: return FindProductQuery async def handle ( self , query : FindProductQuery ) -> OptionalResponse : if query . ref != '123' : raise ProductNotFoundError . create ( detail = { 'ref' : query . ref }) return { 'ref' : query . ref } async def main () -> None : commands_bus = SimpleCommandBus ([ StoreProductCommandHandler ()]) await commands_bus . dispatch ( StoreProductCommand ( '123' )) query_bus = SimpleQueryBus ([ FindProductQueryHandler ()]) response = await query_bus . ask ( FindProductQuery ( '123' )) print ( response ) if __name__ == '__main__' : get_event_loop () . run_until_complete ( main ()) License MIT WIP","title":"aioddd"},{"location":"#async-python-ddd-utilities-library","text":"Key Features: Aggregates : Aggregate & AggregateRoot ValueObjects : Id, Timestamp & StrDateTime CQRS : Command, CommandBus, SimpleCommandBus, Query, Response, QueryHandler, QueryBus & SimpleQueryBus EventSourcing : Event, EventMapper, EventPublisher, EventHandler, EventBus, SimpleEventBus & InternalEventPublisher Errors : raise_, BaseError, NotFoundError, ConflictError, BadRequestError, UnauthorizedError, ForbiddenError, UnknownError, IdInvalidError, TimestampInvalidError, DateTimeInvalidError, EventMapperNotFoundError, EventNotPublishedError, CommandNotRegisteredError & QueryNotRegisteredError Tests : AsyncMock & mock Utils : get_env & get_simple_logger","title":"Async Python DDD utilities library"},{"location":"#requirements","text":"Python 3.6+","title":"Requirements"},{"location":"#installation","text":"python3 -m pip install aioddd","title":"Installation"},{"location":"#example","text":"from asyncio import get_event_loop from dataclasses import dataclass from typing import Type from aioddd import NotFoundError , \\ Command , CommandHandler , SimpleCommandBus , \\ Query , QueryHandler , OptionalResponse , SimpleQueryBus , Event _products = [] class ProductStored ( Event ): @dataclass class Attributes : ref : str attributes : Attributes class StoreProductCommand ( Command ): def __init__ ( self , ref : str ): self . ref = ref class StoreProductCommandHandler ( CommandHandler ): def subscribed_to ( self ) -> Type [ Command ]: return StoreProductCommand async def handle ( self , command : StoreProductCommand ) -> None : _products . append ( command . ref ) class ProductNotFoundError ( NotFoundError ): _code = 'product_not_found' _title = 'Product not found' class FindProductQuery ( Query ): def __init__ ( self , ref : str ): self . ref = ref class FindProductQueryHandler ( QueryHandler ): def subscribed_to ( self ) -> Type [ Query ]: return FindProductQuery async def handle ( self , query : FindProductQuery ) -> OptionalResponse : if query . ref != '123' : raise ProductNotFoundError . create ( detail = { 'ref' : query . ref }) return { 'ref' : query . ref } async def main () -> None : commands_bus = SimpleCommandBus ([ StoreProductCommandHandler ()]) await commands_bus . dispatch ( StoreProductCommand ( '123' )) query_bus = SimpleQueryBus ([ FindProductQueryHandler ()]) response = await query_bus . ask ( FindProductQuery ( '123' )) print ( response ) if __name__ == '__main__' : get_event_loop () . run_until_complete ( main ())","title":"Example"},{"location":"#license","text":"MIT","title":"License"},{"location":"#wip","text":"","title":"WIP"}]}
+{"config":{"indexing":"full","lang":["en"],"min_search_length":3,"prebuild_index":false,"separator":"[\\s\\-]+"},"docs":[{"location":"","text":"Async Python DDD utilities library Key Features: Aggregates : Aggregate & AggregateRoot ValueObjects : Id, Timestamp & StrDateTime CQRS : Command, CommandBus, SimpleCommandBus, Query, Response, QueryHandler, QueryBus & SimpleQueryBus EventSourcing : Event, EventMapper, EventPublisher, EventHandler, EventBus, SimpleEventBus & InternalEventPublisher Errors : raise_, BaseError, NotFoundError, ConflictError, BadRequestError, UnauthorizedError, ForbiddenError, UnknownError, IdInvalidError, TimestampInvalidError, DateTimeInvalidError, EventMapperNotFoundError, EventNotPublishedError, CommandNotRegisteredError & QueryNotRegisteredError Tests : AsyncMock & mock Utils : get_env & get_simple_logger Requirements Python 3.7+ Installation python3 -m pip install aioddd Example from asyncio import get_event_loop from dataclasses import dataclass from typing import Type from aioddd import NotFoundError , \\ Command , CommandHandler , SimpleCommandBus , \\ Query , QueryHandler , OptionalResponse , SimpleQueryBus , Event _products = [] class ProductStored ( Event ): @dataclass class Attributes : ref : str attributes : Attributes class StoreProductCommand ( Command ): def __init__ ( self , ref : str ): self . ref = ref class StoreProductCommandHandler ( CommandHandler ): def subscribed_to ( self ) -> Type [ Command ]: return StoreProductCommand async def handle ( self , command : StoreProductCommand ) -> None : _products . append ( command . ref ) class ProductNotFoundError ( NotFoundError ): _code = 'product_not_found' _title = 'Product not found' class FindProductQuery ( Query ): def __init__ ( self , ref : str ): self . ref = ref class FindProductQueryHandler ( QueryHandler ): def subscribed_to ( self ) -> Type [ Query ]: return FindProductQuery async def handle ( self , query : FindProductQuery ) -> OptionalResponse : if query . ref != '123' : raise ProductNotFoundError . create ( detail = { 'ref' : query . ref }) return { 'ref' : query . ref } async def main () -> None : commands_bus = SimpleCommandBus ([ StoreProductCommandHandler ()]) await commands_bus . dispatch ( StoreProductCommand ( '123' )) query_bus = SimpleQueryBus ([ FindProductQueryHandler ()]) response = await query_bus . ask ( FindProductQuery ( '123' )) print ( response ) if __name__ == '__main__' : get_event_loop () . run_until_complete ( main ()) License MIT WIP","title":"aioddd"},{"location":"#async-python-ddd-utilities-library","text":"Key Features: Aggregates : Aggregate & AggregateRoot ValueObjects : Id, Timestamp & StrDateTime CQRS : Command, CommandBus, SimpleCommandBus, Query, Response, QueryHandler, QueryBus & SimpleQueryBus EventSourcing : Event, EventMapper, EventPublisher, EventHandler, EventBus, SimpleEventBus & InternalEventPublisher Errors : raise_, BaseError, NotFoundError, ConflictError, BadRequestError, UnauthorizedError, ForbiddenError, UnknownError, IdInvalidError, TimestampInvalidError, DateTimeInvalidError, EventMapperNotFoundError, EventNotPublishedError, CommandNotRegisteredError & QueryNotRegisteredError Tests : AsyncMock & mock Utils : get_env & get_simple_logger","title":"Async Python DDD utilities library"},{"location":"#requirements","text":"Python 3.7+","title":"Requirements"},{"location":"#installation","text":"python3 -m pip install aioddd","title":"Installation"},{"location":"#example","text":"from asyncio import get_event_loop from dataclasses import dataclass from typing import Type from aioddd import NotFoundError , \\ Command , CommandHandler , SimpleCommandBus , \\ Query , QueryHandler , OptionalResponse , SimpleQueryBus , Event _products = [] class ProductStored ( Event ): @dataclass class Attributes : ref : str attributes : Attributes class StoreProductCommand ( Command ): def __init__ ( self , ref : str ): self . ref = ref class StoreProductCommandHandler ( CommandHandler ): def subscribed_to ( self ) -> Type [ Command ]: return StoreProductCommand async def handle ( self , command : StoreProductCommand ) -> None : _products . append ( command . ref ) class ProductNotFoundError ( NotFoundError ): _code = 'product_not_found' _title = 'Product not found' class FindProductQuery ( Query ): def __init__ ( self , ref : str ): self . ref = ref class FindProductQueryHandler ( QueryHandler ): def subscribed_to ( self ) -> Type [ Query ]: return FindProductQuery async def handle ( self , query : FindProductQuery ) -> OptionalResponse : if query . ref != '123' : raise ProductNotFoundError . create ( detail = { 'ref' : query . ref }) return { 'ref' : query . ref } async def main () -> None : commands_bus = SimpleCommandBus ([ StoreProductCommandHandler ()]) await commands_bus . dispatch ( StoreProductCommand ( '123' )) query_bus = SimpleQueryBus ([ FindProductQueryHandler ()]) response = await query_bus . ask ( FindProductQuery ( '123' )) print ( response ) if __name__ == '__main__' : get_event_loop () . run_until_complete ( main ())","title":"Example"},{"location":"#license","text":"MIT","title":"License"},{"location":"#wip","text":"","title":"WIP"}]}