Merge branch 'release/4.1.0'

fastapi_template/cli.py

@@ -337,6 +337,21 @@ def checker(ctx: BuilderContext) -> bool:
                 )
             ),
         ),
+        MenuEntry(
+            code="enable_taskiq",
+            cli_name="taskiq",
+            user_view="Add Taskiq support",
+            description=(
+                "{what} is an async task manager.\n"
+                "You can think of taskiq as a {celery}, but async.".format(
+                    what=colored("Taskiq", color="green"),
+                    celery=colored(
+                        "celery",
+                        color="cyan",
+                    ),
+                )
+            ),
+        ),
         MenuEntry(
             code="enable_migrations",
             cli_name="migrations",

fastapi_template/template/cookiecutter.json

@@ -20,6 +20,9 @@
     "enable_migrations": {
         "type": "bool"
     },
+    "enable_taskiq": {
+        "type": "bool"
+    },
     "enable_kube": {
         "type": "bool"
     },

fastapi_template/template/{{cookiecutter.project_name}}/conditional_files.json

@@ -209,5 +209,11 @@
             "{{cookiecutter.project_name}}/db_tortoise/migrations/models/0_20210928165300_init_sqlite.sql",
             "{{cookiecutter.project_name}}/db_tortoise/migrations/models/1_20210928165300_init_dummy_sqlite.sql"
         ]
+    },
+    "Taskiq support":{
+        "enabled": "{{cookiecutter.enable_taskiq}}",
+        "resources": [
+            "{{cookiecutter.project_name}}/taskiq.py"
+        ]
     }
 }

fastapi_template/template/{{cookiecutter.project_name}}/deploy/docker-compose.dev.yml

@@ -11,3 +11,16 @@ services:
     environment:
       # Enables autoreload.
       {{cookiecutter.project_name | upper}}_RELOAD: "True"
+
+  {%- if cookiecutter.enable_taskiq == "True" %}
+
+  taskiq-worker:
+    volumes:
+      # Adds current directory as volume.
+      - .:/app/src/
+    command:
+      - taskiq
+      - worker
+      - {{cookiecutter.project_name}}.taskiq:broker
+      - --reload
+  {%- endif %}

fastapi_template/template/{{cookiecutter.project_name}}/deploy/docker-compose.yml

@@ -1,7 +1,7 @@
 version: '3.9'
 
 services:
-  api:
+  api: &main_app
     build:
       context: .
       dockerfile: ./deploy/Dockerfile
@@ -61,6 +61,9 @@ services:
       {%- if cookiecutter.enable_rmq == 'True' %}
       {{cookiecutter.project_name | upper }}_RABBIT_HOST: {{cookiecutter.project_name}}-rmq
       {%- endif %}
+      {%- if cookiecutter.enable_redis == 'True' %}
+      {{cookiecutter.project_name | upper }}_REDIS_HOST: {{cookiecutter.project_name}}-redis
+      {%- endif %}
       {%- if cookiecutter.enable_kafka == "True" %}
       TESTKAFKA_KAFKA_BOOTSTRAP_SERVERS: '["{{cookiecutter.project_name}}-kafka:9092"]'
       {%- endif %}
@@ -69,6 +72,17 @@ services:
       - {{cookiecutter.project_name}}-db-data:/db_data/
     {%- endif %}
 
+  {%- if cookiecutter.enable_taskiq == "True" %}
+
+  taskiq-worker:
+    <<: *main_app
+    labels: []
+    command:
+      - taskiq
+      - worker
+      - {{cookiecutter.project_name}}.taskiq:broker
+  {%- endif %}
+
   {%- if cookiecutter.db_info.name == "postgresql" %}
 
   db:

fastapi_template/template/{{cookiecutter.project_name}}/pyproject.toml

@@ -140,6 +140,23 @@ asynctest = "^0.13.0"
 nest-asyncio = "^1.5.6"
 {%- endif %}
 httpx = "^0.23.3"
+{%- if cookiecutter.enable_taskiq == "True" %}
+taskiq = "^0"
+taskiq-fastapi = "^0"
+
+    {%- if cookiecutter.enable_redis == "True" %}
+taskiq-redis = "^0"
+    {%- endif %}
+
+    {%- if cookiecutter.enable_rmq == "True" %}
+taskiq-aio-pika = "^0"
+    {%- endif %}
+
+    {%- if (cookiecutter.enable_rmq or cookiecutter.enable_rmq) != "True" %}
+pyzmq = "^25.0.2"
+    {%- endif %}
+
+{%- endif %}
 
 [tool.isort]
 profile = "black"
@@ -177,8 +194,8 @@ filterwarnings = [
     "ignore::DeprecationWarning",
     "ignore:.*unclosed.*:ResourceWarning",
 ]
-{%- if cookiecutter.db_info.name != "none" %}
 env = [
+    "{{cookiecutter.project_name | upper}}_ENVIRONMENT=pytest",
     {%- if cookiecutter.db_info.name == "sqlite" %}
     "{{cookiecutter.project_name | upper}}_DB_FILE=test_db.sqlite3",
     {%- else %}
@@ -191,7 +208,6 @@ env = [
     "{{cookiecutter.project_name | upper}}_SENTRY_DSN=",
     {%- endif %}
 ]
-{%- endif %}
 
 {%- if cookiecutter.orm == "tortoise" %}
 {%- if cookiecutter.enable_migrations == "True" %}

fastapi_template/template/{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/db_psycopg/dependencies.py

@@ -1,8 +1,11 @@
 from psycopg_pool import AsyncConnectionPool
 from starlette.requests import Request
 
+{%- if cookiecutter.enable_taskiq == "True" %}
+from taskiq import TaskiqDepends
+{%- endif %}
 
-async def get_db_pool(request: Request) -> AsyncConnectionPool:
+async def get_db_pool(request: Request {%- if cookiecutter.enable_taskiq == "True" %} = TaskiqDepends(){%- endif %}) -> AsyncConnectionPool:
     """
     Return database connections pool.
 

fastapi_template/template/{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/db_sa/dependencies.py

@@ -3,8 +3,12 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from starlette.requests import Request
 
+{%- if cookiecutter.enable_taskiq == "True" %}
+from taskiq import TaskiqDepends
+{%- endif %}
 
-async def get_db_session(request: Request) -> AsyncGenerator[AsyncSession, None]:
+
+async def get_db_session(request: Request {%- if cookiecutter.enable_taskiq == "True" %} = TaskiqDepends(){%- endif %}) -> AsyncGenerator[AsyncSession, None]:
     """
     Create and get database session.
 

fastapi_template/template/{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/logging.py

@@ -83,16 +83,19 @@ def record_formatter(record: dict[str, Any]) -> str:  # pragma: no cover
 
 def configure_logging() -> None:  # pragma: no cover
     """Configures logging."""
-    loggers = (
-        logging.getLogger(name)
-        for name in logging.root.manager.loggerDict
-        if name.startswith("uvicorn.")
-    )
-    for uvicorn_logger in loggers:
-        uvicorn_logger.handlers = []
+    intercept_handler = InterceptHandler()
+
+    logging.basicConfig(handlers=[intercept_handler], level=logging.NOTSET)
+
+    for logger_name in logging.root.manager.loggerDict:
+        if logger_name.startswith("uvicorn."):
+            logging.getLogger(logger_name).handlers = []
+        {%- if cookiecutter.enable_taskiq == "True" %}
+        if logger_name.startswith("taskiq."):
+            logging.getLogger(logger_name).root.handlers = [intercept_handler]
+        {%- endif %}
 
     # change handler for default uvicorn logger
-    intercept_handler = InterceptHandler()
     logging.getLogger("uvicorn").handlers = [intercept_handler]
     logging.getLogger("uvicorn.access").handlers = [intercept_handler]
 

fastapi_template/template/{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/services/kafka/dependencies.py

@@ -1,8 +1,12 @@
 from aiokafka import AIOKafkaProducer
 from fastapi import Request
 
+{%- if cookiecutter.enable_taskiq == "True" %}
+from taskiq import TaskiqDepends
+{%- endif %}
 
-def get_kafka_producer(request: Request) -> AIOKafkaProducer:  # pragma: no cover
+
+def get_kafka_producer(request: Request {%- if cookiecutter.enable_taskiq == "True" %} = TaskiqDepends(){%- endif %}) -> AIOKafkaProducer:  # pragma: no cover
     """
     Returns kafka producer.
 

fastapi_template/template/{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/services/rabbit/dependencies.py

@@ -2,8 +2,12 @@
 from aio_pika.pool import Pool
 from fastapi import Request
 
+{%- if cookiecutter.enable_taskiq == "True" %}
+from taskiq import TaskiqDepends
+{%- endif %}
 
-def get_rmq_channel_pool(request: Request) -> Pool[Channel]:  # pragma: no cover
+
+def get_rmq_channel_pool(request: Request {%- if cookiecutter.enable_taskiq == "True" %} = TaskiqDepends(){%- endif %}) -> Pool[Channel]:  # pragma: no cover
     """
     Get channel pool from the state.
 

fastapi_template/template/{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/services/redis/dependency.py

@@ -3,8 +3,12 @@
 from redis.asyncio import Redis
 from starlette.requests import Request
 
+{%- if cookiecutter.enable_taskiq == "True" %}
+from taskiq import TaskiqDepends
+{%- endif %}
 
-async def get_redis_pool(request: Request) -> AsyncGenerator[Redis, None]:  # pragma: no cover
+
+async def get_redis_pool(request: Request {%- if cookiecutter.enable_taskiq == "True" %} = TaskiqDepends(){%- endif %}) -> AsyncGenerator[Redis, None]:  # pragma: no cover
     """
     Returns connection pool.
 

fastapi_template/template/{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/taskiq.py

@@ -0,0 +1,37 @@
+from {{cookiecutter.project_name}}.settings import settings
+
+import taskiq_fastapi
+from taskiq import TaskiqState, ZeroMQBroker, InMemoryBroker, TaskiqEvents
+
+{%- if cookiecutter.enable_redis == "True" %}
+from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend
+{%- endif %}
+
+{%- if cookiecutter.enable_rmq == "True" %}
+from taskiq_aio_pika import AioPikaBroker
+{%- endif %}
+
+{%- if cookiecutter.enable_loguru %}
+from {{cookiecutter.project_name}}.logging import configure_logging
+{%- endif %}
+
+
+{%- if cookiecutter.enable_redis == "True" %}
+result_backend = RedisAsyncResultBackend(
+    redis_url=str(settings.redis_url.with_path("/1")),
+)
+{%- endif %}
+
+
+{%- if cookiecutter.enable_rmq == "True" %}
+broker = AioPikaBroker(str(settings.rabbit_url), {%- if cookiecutter.enable_redis == "True" %}result_backend=result_backend, {%- endif %})
+{%- elif cookiecutter.enable_redis == "True" %}
+broker = ListQueueBroker(str(settings.redis_url.with_path("/1")), result_backend=result_backend,)
+{%- else %}
+broker = ZeroMQBroker()
+{%- endif %}
+
+if settings.environment.lower() == "pytest":
+    broker = InMemoryBroker()
+
+taskiq_fastapi.init(broker, "tkiqtest.web.application:get_app")

fastapi_template/template/{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/web/lifetime.py

@@ -20,6 +20,10 @@
 from {{cookiecutter.project_name}}.services.kafka.lifetime import init_kafka, shutdown_kafka
 {%- endif %}
 
+{%- if cookiecutter.enable_taskiq == "True" %}
+from {{cookiecutter.project_name}}.taskiq import broker
+{%- endif %}
+
 
 {%- if cookiecutter.orm == "ormar" %}
 from {{cookiecutter.project_name}}.db.config import database
@@ -255,6 +259,10 @@ def register_startup_event(app: FastAPI) -> Callable[[], Awaitable[None]]:  # pr
 
     @app.on_event("startup")
     async def _startup() -> None:  # noqa: WPS430
+        {%- if cookiecutter.enable_taskiq == "True" %}
+        if not broker.is_worker_process:
+            await broker.startup()
+        {%- endif %}
         {%- if cookiecutter.orm == "sqlalchemy" %}
         _setup_db(app)
         {%- elif cookiecutter.orm == "ormar" %}
@@ -297,6 +305,10 @@ def register_shutdown_event(app: FastAPI) -> Callable[[], Awaitable[None]]:  # p
 
     @app.on_event("shutdown")
     async def _shutdown() -> None:  # noqa: WPS430
+        {%- if cookiecutter.enable_taskiq == "True" %}
+        if not broker.is_worker_process:
+            await broker.shutdown()
+        {%- endif %}
         {%- if cookiecutter.orm == "sqlalchemy" %}
         await app.state.db_engine.dispose()
         {% elif cookiecutter.orm == "ormar" %}

fastapi_template/tests/conftest.py

@@ -57,6 +57,7 @@ def default_context(project_name: str) -> None:
         db="none",
         db_info=Database(name="none").dict(),
         enable_redis=False,
+        enable_taskiq=False,
         enable_migrations=False,
         enable_kube=False,
         enable_routers=True,

fastapi_template/tests/test_generator.py

@@ -144,6 +144,7 @@ def test_without_dummy(default_context: BuilderContext, orm: str):
 )
 def test_redis(default_context: BuilderContext, api: str):
     default_context.enable_redis = True
+    default_context.enable_taskiq = True
     default_context.api_type = api
     run_default_check(default_context)
 
@@ -157,6 +158,7 @@ def test_redis(default_context: BuilderContext, api: str):
 )
 def test_rmq(default_context: BuilderContext, api: str):
     default_context.enable_rmq = True
+    default_context.enable_taskiq = True
     default_context.api_type = api
     run_default_check(default_context)
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "fastapi_template"
-version = "4.0.4"
+version = "4.1.0"
 description = "Feature-rich robust FastAPI template"
 authors = ["Pavel Kirilin <win10@list.ru>"]
 packages = [{ include = "fastapi_template" }]