From d9e5effdce09ae46a86f2c5ecbd89d2e5dad9f3e Mon Sep 17 00:00:00 2001 From: Viktor Smagin Date: Wed, 8 Jul 2026 23:01:55 +0400 Subject: [PATCH] Skeleton ready --- .editorconfig | 23 + .env.example | 13 + .gitignore | 15 + Jenkinsfile | 35 ++ LICENSE | 21 + Makefile | 14 + README.md | 405 ++++++++++++++++++ deploy/local/compose.yaml | 57 +++ deploy/local/python.Dockerfile | 11 + deploy/main/compose.yaml | 77 ++++ deploy/main/python.Dockerfile | 13 + entry/log_user_messages.py | 37 ++ examples/message.json | 6 + feature/__init__.py | 0 feature/listen_user_message/__init__.py | 0 .../listen_user_message.py | 15 + .../listen_user_message/manager/__init__.py | 0 .../listen_user_message/manager/message.py | 37 ++ feature/log_bus_message/__init__.py | 0 feature/log_bus_message/listener/__init__.py | 0 .../log_bus_message/listener/log_message.py | 25 ++ feature/log_bus_message/log_bus_message.py | 18 + kernel/__init__.py | 0 kernel/application/__init__.py | 0 kernel/application/application.py | 39 ++ kernel/application/enum/__init__.py | 0 kernel/application/enum/date_time.py | 4 + kernel/application/enum/encoding.py | 4 + kernel/application/kernel.py | 5 + kernel/application/manager/__init__.py | 0 kernel/application/manager/stop_event.py | 13 + kernel/application/stop_event.py | 4 + kernel/application/view/__init__.py | 0 kernel/application/view/application.py | 7 + kernel/application/view/configuration.py | 28 ++ kernel/application/view/rabbitmq.py | 11 + kernel/application/view/redis.py | 8 + kernel/event/__init__.py | 0 kernel/event/listener.py | 16 + kernel/event/registrar.py | 25 ++ kernel/feature/__init__.py | 0 kernel/feature/provider.py | 17 + kernel/rabbitmq/__init__.py | 0 kernel/rabbitmq/client.py | 144 +++++++ kernel/rabbitmq/enum/__init__.py | 0 kernel/rabbitmq/enum/content_type.py | 5 + kernel/rabbitmq/exception/__init__.py | 0 .../exception/connection_not_established.py | 7 + .../rabbitmq/exception/connection_timeout.py | 7 + .../rabbitmq/exception/rabbitmq_exception.py | 5 + kernel/rabbitmq/view/__init__.py | 0 kernel/rabbitmq/view/authentication_params.py | 7 + kernel/rabbitmq/view/connection_params.py | 8 + requirements.txt | 9 + shared/__init__.py | 0 shared/application/__init__.py | 0 shared/application/factory/__init__.py | 0 shared/application/factory/application.py | 10 + shared/application/factory/logger.py | 27 ++ shared/application/manager/__init__.py | 0 .../manager/listener_arguments_parser.py | 40 ++ shared/application/rabbitmq_application.py | 56 +++ shared/application/rabbitmq_kernel.py | 139 ++++++ shared/application/service/__init__.py | 0 .../application/service/environment_loader.py | 29 ++ shared/application/view/__init__.py | 0 shared/application/view/listener_arguments.py | 8 + shared/bus/__init__.py | 0 shared/bus/view/__init__.py | 0 shared/bus/view/message.py | 9 + shared/event/__init__.py | 0 shared/event/type.py | 5 + typings/aioredis/__init__.pyi | 321 ++++++++++++++ typings/aioredlock/__init__.pyi | 95 ++++ typings/punq/__init__.pyi | 55 +++ 75 files changed, 1989 insertions(+) create mode 100644 .editorconfig create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 Jenkinsfile create mode 100644 LICENSE create mode 100644 Makefile create mode 100644 README.md create mode 100644 deploy/local/compose.yaml create mode 100644 deploy/local/python.Dockerfile create mode 100644 deploy/main/compose.yaml create mode 100644 deploy/main/python.Dockerfile create mode 100644 entry/log_user_messages.py create mode 100644 examples/message.json create mode 100644 feature/__init__.py create mode 100644 feature/listen_user_message/__init__.py create mode 100644 feature/listen_user_message/listen_user_message.py create mode 100644 feature/listen_user_message/manager/__init__.py create mode 100644 feature/listen_user_message/manager/message.py create mode 100644 feature/log_bus_message/__init__.py create mode 100644 feature/log_bus_message/listener/__init__.py create mode 100644 feature/log_bus_message/listener/log_message.py create mode 100644 feature/log_bus_message/log_bus_message.py create mode 100644 kernel/__init__.py create mode 100644 kernel/application/__init__.py create mode 100644 kernel/application/application.py create mode 100644 kernel/application/enum/__init__.py create mode 100644 kernel/application/enum/date_time.py create mode 100644 kernel/application/enum/encoding.py create mode 100644 kernel/application/kernel.py create mode 100644 kernel/application/manager/__init__.py create mode 100644 kernel/application/manager/stop_event.py create mode 100644 kernel/application/stop_event.py create mode 100644 kernel/application/view/__init__.py create mode 100644 kernel/application/view/application.py create mode 100644 kernel/application/view/configuration.py create mode 100644 kernel/application/view/rabbitmq.py create mode 100644 kernel/application/view/redis.py create mode 100644 kernel/event/__init__.py create mode 100644 kernel/event/listener.py create mode 100644 kernel/event/registrar.py create mode 100644 kernel/feature/__init__.py create mode 100644 kernel/feature/provider.py create mode 100644 kernel/rabbitmq/__init__.py create mode 100644 kernel/rabbitmq/client.py create mode 100644 kernel/rabbitmq/enum/__init__.py create mode 100644 kernel/rabbitmq/enum/content_type.py create mode 100644 kernel/rabbitmq/exception/__init__.py create mode 100644 kernel/rabbitmq/exception/connection_not_established.py create mode 100644 kernel/rabbitmq/exception/connection_timeout.py create mode 100644 kernel/rabbitmq/exception/rabbitmq_exception.py create mode 100644 kernel/rabbitmq/view/__init__.py create mode 100644 kernel/rabbitmq/view/authentication_params.py create mode 100644 kernel/rabbitmq/view/connection_params.py create mode 100644 requirements.txt create mode 100644 shared/__init__.py create mode 100644 shared/application/__init__.py create mode 100644 shared/application/factory/__init__.py create mode 100644 shared/application/factory/application.py create mode 100644 shared/application/factory/logger.py create mode 100644 shared/application/manager/__init__.py create mode 100644 shared/application/manager/listener_arguments_parser.py create mode 100644 shared/application/rabbitmq_application.py create mode 100644 shared/application/rabbitmq_kernel.py create mode 100644 shared/application/service/__init__.py create mode 100644 shared/application/service/environment_loader.py create mode 100644 shared/application/view/__init__.py create mode 100644 shared/application/view/listener_arguments.py create mode 100644 shared/bus/__init__.py create mode 100644 shared/bus/view/__init__.py create mode 100644 shared/bus/view/message.py create mode 100644 shared/event/__init__.py create mode 100644 shared/event/type.py create mode 100644 typings/aioredis/__init__.pyi create mode 100644 typings/aioredlock/__init__.pyi create mode 100644 typings/punq/__init__.pyi diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..732a591 --- /dev/null +++ b/.editorconfig @@ -0,0 +1,23 @@ +root = true + +[*] +charset = utf-8 +end_of_line = lf +insert_final_newline = true +trim_trailing_whitespace = true + +[*.py] +indent_style = space +indent_size = 4 +max_line_length = 88 +insert_final_newline = true + +[*.{yml,yaml,json,md}] +indent_style = space +indent_size = 2 + +[*.md] +trim_trailing_whitespace = false + +[Makefile] +indent_style = tab diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..f1fafc8 --- /dev/null +++ b/.env.example @@ -0,0 +1,13 @@ +APP_NAME=Skeleton +APP_ENV=local + +RABBITMQ_HOST=localhost +RABBITMQ_PORT=5672 +RABBITMQ_USER=guest +RABBITMQ_PASS=guest +RABBITMQ_TIMEOUT=10 +RABBITMQ_VHOST=/ + +REDIS_HOST=application-skeleton-redis +REDIS_PORT=6379 +REDIS_DB=0 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..79497d4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,15 @@ +.venv/ +.vscode/ +.idea/ +*.pyc +*.pyo +*.pyd +__pycache__/ +*.so +*.egg +*.egg-info +dist/ +build/ +*.log +*.env +pyrightconfig.json diff --git a/Jenkinsfile b/Jenkinsfile new file mode 100644 index 0000000..c1f80fb --- /dev/null +++ b/Jenkinsfile @@ -0,0 +1,35 @@ +pipeline { + agent { + label "backend-agent" + } + + environment { + PROJECT_NAME = "application-skeleton" + PROJECT_CREDENTIALS = credentials("applicationSkeleton.credentials") + } + + stages { + stage("Doing something") { + steps() { + script { + pipe = setupWebhookPipeline() + + def dateTag = sh( + script: 'date -u +%Y%m%d%H%M%S', + returnStdout: true + ).trim() + + env.VERSION = "v1.2.3-${dateTag}" + } + } + } + } + post { + success { + + } + unsuccessful { + + } + } +} diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..b86210a --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Diffhead + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..0018abd --- /dev/null +++ b/Makefile @@ -0,0 +1,14 @@ +Compose = docker compose --env-file .env -f deploy/local/compose.yaml + +build: + @$(Compose) build + @$(Compose) up -d +start: + @$(Compose) start +stop: + @$(Compose) stop +restart: + @$(Compose) restart +delete: + @$(Compose) down + diff --git a/README.md b/README.md new file mode 100644 index 0000000..b6272bb --- /dev/null +++ b/README.md @@ -0,0 +1,405 @@ +# Feature-Oriented Event-Driven Application Skeleton + +A Python-based asynchronous application framework designed for building scalable, event-driven systems using a feature-oriented architecture. This skeleton provides a solid foundation for building complex applications with clear separation of concerns, dependency injection, and event-driven communication patterns. + +## Overview + +This project implements a feature-oriented architecture where functionality is organized into independent, self-contained features that communicate through an event-driven model. Each feature is responsible for its own initialization, configuration, and bootstrap logic, making the system highly modular and testable. + +### Key Concepts + +- **Feature-Oriented**: Application logic is divided into independent features, each with its own provider for configuration and lifecycle management +- **Event-Driven**: Features communicate asynchronously through an event emitter +- **Async-First**: Built on Python's `asyncio` for high-performance concurrent operations +- **Dependency Injection**: Uses the `punq` container for loose coupling and testability +- **Kernel-Centric**: Core infrastructure (application lifecycle, messaging, events) lives in the `kernel` namespace + +## Project Structure + +``` +FeatureOrientedSkeleton/ +├── kernel/ # Core framework and infrastructure +│ ├── application/ # Application lifecycle management +│ │ ├── application.py # Abstract base application class +│ │ ├── kernel.py # Abstract kernel (initialization point) +│ │ ├── stop_event.py # Graceful shutdown signaling +│ │ ├── enum/ # Core enumerations +│ │ └── manager/ # Event managers +│ ├── event/ # Event handling infrastructure +│ │ ├── listener.py # Base listener interface +│ │ └── registrar.py # Event listener registration mixin +│ ├── feature/ # Feature provider interface +│ │ └── provider.py # Abstract feature provider +│ └── rabbitmq/ # RabbitMQ client and connection +│ ├── client.py # RabbitMQ messaging client +│ ├── enum/ # RabbitMQ-specific types +│ └── exception/ # RabbitMQ exceptions +│ +├── shared/ # Shared utilities and implementations +│ ├── application/ # Concrete application implementations +│ │ ├── rabbitmq_application.py # RabbitMQ-based application +│ │ ├── rabbitmq_kernel.py # RabbitMQ kernel with DI setup +│ │ └── factory/ # Factory patterns for common objects +│ ├── event/ # Shared event types and structures +│ ├── service/ # Cross-cutting services +│ ├── manager/ # Utilities (argument parsing, etc) +│ └── view/ # View models and configurations +│ +├── feature/ # Application features (domain logic) +│ ├── listen_rabbitmq/ # Feature: Listen to RabbitMQ messages +│ │ ├── listen_rabbitmq.py # Feature provider +│ │ ├── consumer/ # Message consumption logic +│ │ ├── service/ # Business logic and services +│ │ └── type/ # Domain types and models +│ │ +│ ├── log_bus_message/ # Feature: Log received messages +│ │ ├── log_bus_message.py # Feature provider +│ │ └── listener/ # Event listeners +│ │ +│ ├── prepare_error_notification/ # Feature: Prepare error notifications +│ ├── send_telegram_notification/ # Feature: Send Telegram notifications +│ └── [other-features]/ # Additional features +│ +├── entry/ # Application entry points +│ └── log_user_messages.py # CLI entry script +│ +├── typings/ # Type stubs for third-party libraries +│ ├── aioredis/ +│ ├── aioredlock/ +│ └── punq/ +│ +├── deploy/ # Deployment configurations +│ ├── local/ # Local development (Docker Compose) +│ │ ├── compose.yaml +│ │ └── python.Dockerfile +│ └── main/ # Production deployment +│ ├── compose.yaml +│ └── python.Dockerfile +│ +├── examples/ # Example payloads and usage +├── Makefile # Development commands +├── pyrightconfig.json # Type checking configuration +├── requirements.txt # Python dependencies +├── .env.example # Environment variables template +└── README.md # This file +``` + +## Architecture Patterns + +### 1. Feature Provider Pattern + +Each feature must implement a **Feature Provider** - a class that inherits from `Provider` (and optionally `Registrar` mixin): + +```python +from typing import override +from kernel.feature.provider import Provider +from kernel.event.registrar import Registrar +from shared.event.type import Type as EventType + +class MyFeature(Provider, Registrar): + @override + def register(self) -> None: + """ + Called during DI container initialization. + - Register classes in the container + - Load configurations + - Set up initial state + """ + self._container.register(MyService) + + @override + def bootstrap(self) -> None: + """ + Called after all features are registered. + - Start background tasks + - Attach event listeners + - Initialize runtime state + """ + self.add_event_listener(EventType.MESSAGE_RECEIVED, MyEventListener) +``` + +### 2. Event Listener Pattern + +Create an event listener that inherits from `Listener[T]`: + +```python +from kernel.event.listener import Listener +from shared.event.type import Type as EventType + +class MyEventListener(Listener[MyEventType]): + async def handle(self, event: str, payload: MyEventType) -> None: + # Process the event + pass +``` + +Attach it in your feature's `bootstrap()` method using the `Registrar` mixin. + +### 3. Dependency Injection + +All dependencies are resolved through the `punq` Container: + +```python +from punq import Container + +class MyService: + def __init__(self, container: Container) -> None: + self._container = container + + def do_something(self) -> None: + dependency = self._container.resolve(SomeDependency) +``` + +## Quick Start + +### 1. Clone and Setup + +```bash +git clone +cd FeatureOrientedSkeleton +cp .env.example .env +``` + +### 2. Using Makefile (Recommended) + +```bash +# Build and start all services +make up + +# View application logs +make logs + +# Stop services +make stop + +# Clean up containers and volumes +make clean +``` + +### 3. Manual Docker Compose + +```bash +# Start services +docker-compose -f deploy/local/compose.yaml up + +# Stop services +docker-compose -f deploy/local/compose.yaml down +``` + +## Running the Example + +The project includes an example application that listens to RabbitMQ messages and logs them: + +```bash +make up +# Application starts and listens on queue 'application.skeleton' +# Send messages and view logs +make logs +``` + +## Creating a New Feature + +### Step 1: Create Feature Directory + +```bash +mkdir -p feature/my_feature/{service,listener,type,consumer} +touch feature/my_feature/__init__.py +``` + +### Step 2: Create Feature Provider + +```python +# feature/my_feature/my_feature.py +from typing import override +from kernel.feature.provider import Provider +from kernel.event.registrar import Registrar + +class MyFeature(Provider, Registrar): + @override + def register(self) -> None: + # Register dependencies and services + pass + + @override + def bootstrap(self) -> None: + # Attach event listeners if needed + pass +``` + +### Step 3: Create Event Listener (if needed) + +```python +# feature/my_feature/listener/my_listener.py +from kernel.event.listener import Listener + +class MyListener(Listener[MyEventType]): + async def handle(self, event: str, payload: MyEventType) -> None: + # Handle the event + pass +``` + +### Step 4: Use in Entry Script + +```python +# entry/my_listener.py +import asyncio +from shared.application.factory.application import Application as ApplicationFactory +from feature.my_feature import MyFeature + +if __name__ == "__main__": + factory = ApplicationFactory() + + application = factory.create_rabbitmq([ + MyFeature, + ]) + + asyncio.run(application.launch(main)) +``` + +## Message Format + +Messages must conform to a specific JSON format: + +```json +{ + "service": "service_name", + "environment": "production", + "exception": "ErrorType", + "message": "Error message text", + "type": "error", + "datetime": "2024-01-01T12:00:00" +} +``` + +See `examples/` directory for sample messages. + +## Environment Configuration + +Copy `.env.example` to `.env` and configure: + +```env +# Application +APP_NAME=application-skeleton +APP_ENV=local + +# RabbitMQ +RABBITMQ_HOST=rabbitmq +RABBITMQ_PORT=5672 +RABBITMQ_USER=guest +RABBITMQ_PASS=guest +RABBITMQ_VHOST=/ +RABBITMQ_TIMEOUT=10 + +# Redis +REDIS_HOST=redis +REDIS_PORT=6379 +REDIS_DB=0 +``` + +## Development + +### Makefile Commands + +```bash +make build # Build Docker images +make up # Start services (builds if needed) +make down # Stop services +make restart # Restart services +make logs # View application logs +make stop # Stop and remove containers +make clean # Remove all containers and volumes +``` + +### Type Checking + +```bash +pyright +``` + +### Code Style + +Follow PEP 8 conventions. Configuration in `.editorconfig`. + +## Key Design Principles + +1. **Kernel is Core**: Infrastructure lives in `kernel/`, never import from `feature/` into `kernel/` +2. **Shared for Common Code**: Cross-cutting concerns and shared implementations belong in `shared/` +3. **Features are Independent**: Features should not depend on other features directly +4. **DI Over Coupling**: Use the DI container to manage dependencies +5. **Async-First**: Embrace async/await throughout the codebase +6. **Events for Communication**: Features communicate through events, not direct calls +7. **Provider Pattern**: Each feature must implement the Provider pattern with `register()` and `bootstrap()` methods + +## Performance Considerations + +- **Graceful Shutdown**: Application responds to SIGTERM and SIGINT signals +- **Resource Cleanup**: Set `stop_grace_period: 3s` in Docker Compose for clean shutdown +- **Connection Pooling**: RabbitMQ and Redis connections are pooled for efficiency +- **Event-Driven**: Non-blocking event handling allows processing many messages concurrently + +## Troubleshooting + +### Application doesn't stop gracefully + +Ensure `stop_grace_period` is set in `docker-compose.yaml`: + +```yaml +services: + application: + stop_grace_period: 3s +``` + +### Messages not being processed + +1. Check RabbitMQ is healthy: `docker ps` +2. Verify queue and routing key match the listeners +3. View logs: `make logs` +4. Check environment variables in `.env` + +### Type checking errors + +Ensure typings are installed: + +```bash +ls typings/ +# Should contain stubs for aioredis, aioredlock, punq +``` + +### ConnectionError to RabbitMQ + +1. Verify `depends_on` conditions in compose.yaml: + ```yaml + depends_on: + rabbitmq: + condition: service_healthy + ``` +2. Check healthcheck passes: `docker ps` +3. Verify credentials in `.env` + +## Dependencies + +Core dependencies include: + +- `punq` - Dependency injection container +- `aio-pika` - RabbitMQ client +- `redis` - Redis client +- `aioredlock` - Distributed locking +- `pyee` - Event emitter + +See `requirements.txt` for complete list and versions. + +## Contributing + +1. Follow the feature-oriented structure +2. Each feature should be self-contained +3. Use dependency injection for all dependencies +4. Add type annotations to all functions +5. Write docstrings for public APIs +6. Test features in isolation using the DI container + +## License + +[Add your license here] + +--- + +**For more information on feature-oriented architecture and event-driven systems, see the examples in the `examples/` directory.** diff --git a/deploy/local/compose.yaml b/deploy/local/compose.yaml new file mode 100644 index 0000000..fb929c0 --- /dev/null +++ b/deploy/local/compose.yaml @@ -0,0 +1,57 @@ +name: application-skeleton + +services: + application: + build: + context: ../../ + dockerfile: deploy/local/python.Dockerfile + container_name: application-skeleton-application + command: [ + "python", "-m", "entry.log_user_messages", + "--queue=application.skeleton", + "--exchange=amq.topic", + "--routing-key=user.message.*" + ] + stop_grace_period: 3s + depends_on: + rabbitmq: + condition: service_healthy + redis: + condition: service_healthy + volumes: + - ../../:/opt/app + networks: + - application-skeleton-network + + rabbitmq: + image: rabbitmq:4.3-management + container_name: application-skeleton-rabbitmq + environment: + RABBITMQ_DEFAULT_USER: ${RABBITMQ_USER} + RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASS} + healthcheck: + test: rabbitmq-diagnostics ping + interval: 3s + timeout: 5s + retries: 10 + start_period: 10s + networks: + - application-skeleton-network + ports: + - "15672:15672" + + redis: + image: redis:8-alpine + container_name: application-skeleton-redis + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 3s + timeout: 5s + retries: 10 + start_period: 10s + networks: + - application-skeleton-network + +networks: + application-skeleton-network: + external: false diff --git a/deploy/local/python.Dockerfile b/deploy/local/python.Dockerfile new file mode 100644 index 0000000..0f257e2 --- /dev/null +++ b/deploy/local/python.Dockerfile @@ -0,0 +1,11 @@ +FROM python:3.12-slim + +ENV PYTHONPATH=/opt/app + +RUN pip install --upgrade pip + +COPY requirements.txt . + +RUN pip install -r requirements.txt + +WORKDIR /opt/app diff --git a/deploy/main/compose.yaml b/deploy/main/compose.yaml new file mode 100644 index 0000000..87dfed4 --- /dev/null +++ b/deploy/main/compose.yaml @@ -0,0 +1,77 @@ +name: application-skeleton + +services: + application: + build: + context: ../../ + dockerfile: deploy/main/python.Dockerfile + image: ${APP_IMAGE} + container_name: application-skeleton-application + depends_on: + rabbitmq: + condition: service_healthy + redis: + condition: service_healthy + command: [ + "python", "-m", "entry.log_user_messages", + "--queue=application.skeleton", + "--exchange=amq.topic", + "--routing-key=user.message.*" + ] + stop_grace_period: 3s + volumes: + - application-skeleton-data:/opt/app/data + environment: + - APP_NAME=${APP_NAME} + - APP_ENV=${APP_ENV} + - RABBITMQ_HOST=${RABBITMQ_HOST} + - RABBITMQ_PORT=${RABBITMQ_PORT} + - RABBITMQ_USER=${RABBITMQ_USER} + - RABBITMQ_PASS=${RABBITMQ_PASS} + - RABBITMQ_TIMEOUT=${RABBITMQ_TIMEOUT} + - RABBITMQ_VHOST=${RABBITMQ_VHOST} + - REDIS_HOST=${REDIS_HOST} + - REDIS_PORT=${REDIS_PORT} + - REDIS_DB=${REDIS_DB} + networks: + - application-skeleton-network + restart: unless-stopped + + rabbitmq: + image: rabbitmq:4.3-management + container_name: application-skeleton-rabbitmq + environment: + RABBITMQ_DEFAULT_USER: ${RABBITMQ_USER} + RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASS} + healthcheck: + test: rabbitmq-diagnostics ping + interval: 3s + timeout: 5s + retries: 10 + start_period: 10s + networks: + - application-skeleton-network + ports: + - "15672:15672" + + redis: + image: redis:8-alpine + container_name: application-skeleton-redis + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 3s + timeout: 5s + retries: 10 + start_period: 10s + networks: + - application-skeleton-network + restart: unless-stopped + +networks: + application-skeleton-network: + external: false + +volumes: + application-skeleton-data: + external: false + diff --git a/deploy/main/python.Dockerfile b/deploy/main/python.Dockerfile new file mode 100644 index 0000000..a60396f --- /dev/null +++ b/deploy/main/python.Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.12-slim + +ENV PYTHONPATH=/opt/app + +RUN pip install --upgrade pip + +COPY requirements.txt . + +RUN pip install -r requirements.txt + +COPY . /opt/app + +WORKDIR /opt/app diff --git a/entry/log_user_messages.py b/entry/log_user_messages.py new file mode 100644 index 0000000..3f380db --- /dev/null +++ b/entry/log_user_messages.py @@ -0,0 +1,37 @@ +import asyncio + +from punq import Container +from aio_pika.abc import AbstractIncomingMessage +from pyee import EventEmitter + +from shared.application.factory.application import Application as ApplicationFactory +from shared.event.type import Type as EventType + +from feature.listen_user_message.listen_user_message import ListenUserMessage +from feature.log_bus_message.log_bus_message import LogBusMessage + +from feature.listen_user_message.manager.message import Message as MessageManager + + +async def main(container: Container, message: AbstractIncomingMessage) -> None: + manager = container.resolve(MessageManager, message=message) + emitter = container.resolve(EventEmitter) + event = EventType.MESSAGE_RECEIVED + + try: + received = manager.get_message() + emitter.emit(event, received) + + await message.ack() + except Exception: + await message.reject() + +if __name__ == "__main__": + factory = ApplicationFactory() + + application = factory.create_rabbitmq([ + ListenUserMessage, + LogBusMessage + ]) + + asyncio.run(application.launch(main)) diff --git a/examples/message.json b/examples/message.json new file mode 100644 index 0000000..6ed4288 --- /dev/null +++ b/examples/message.json @@ -0,0 +1,6 @@ +{ + "id":"019f42ff-de59-7a17-8189-a36b68191a71", + "source":"019f42ff-dff9-7b68-832f-a6323a078105", + "type":"user.message", + "message":"Hello my friend, i'm there" +} diff --git a/feature/__init__.py b/feature/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/feature/listen_user_message/__init__.py b/feature/listen_user_message/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/feature/listen_user_message/listen_user_message.py b/feature/listen_user_message/listen_user_message.py new file mode 100644 index 0000000..96691ac --- /dev/null +++ b/feature/listen_user_message/listen_user_message.py @@ -0,0 +1,15 @@ +from typing import override + +from kernel.feature.provider import Provider +from kernel.event.registrar import Registrar + +from feature.listen_user_message.manager.message import Message + +class ListenUserMessage(Provider, Registrar): + @override + def register(self) -> None: + self._container.register(Message, Message) + + @override + def bootstrap(self) -> None: + pass diff --git a/feature/listen_user_message/manager/__init__.py b/feature/listen_user_message/manager/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/feature/listen_user_message/manager/message.py b/feature/listen_user_message/manager/message.py new file mode 100644 index 0000000..0f4892c --- /dev/null +++ b/feature/listen_user_message/manager/message.py @@ -0,0 +1,37 @@ +from aio_pika.abc import AbstractIncomingMessage +from json import loads as json_loads + +from kernel.application.enum.encoding import Encoding + +from shared.bus.view.message import Message as BusMessage + + +class Message: + __message: AbstractIncomingMessage + __required_columns = [ + "id", + "type", + "source", + "message", + ] + + def __init__(self, message: AbstractIncomingMessage) -> None: + self.__message = message + + def get_message(self) -> BusMessage: + payload: str = self.__message.body.decode(Encoding.utf8) + payload_json: dict = json_loads(payload) + + for column in self.__required_columns: + if column not in payload_json: + raise ValueError( + f"Missing required column: {column}" + ) + + return BusMessage( + id=payload_json["id"], + type=payload_json["type"], + source=payload_json["source"], + message=payload_json["message"] + ) + diff --git a/feature/log_bus_message/__init__.py b/feature/log_bus_message/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/feature/log_bus_message/listener/__init__.py b/feature/log_bus_message/listener/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/feature/log_bus_message/listener/log_message.py b/feature/log_bus_message/listener/log_message.py new file mode 100644 index 0000000..45cc233 --- /dev/null +++ b/feature/log_bus_message/listener/log_message.py @@ -0,0 +1,25 @@ +from typing import override +from logging import Logger + +from kernel.event.listener import Listener + +from shared.bus.view.message import Message + + +class LogMessage(Listener[Message]): + @override + async def handle(self, event: str, payload: Message) -> None: + logger: Logger = self._container.resolve(Logger) + + logger.info(f"Event: {event}\n") + + content = ( + f"Received message\n" + f"Id: {payload.id}\n" + f"Type: {payload.type}\n" + f"Source: {payload.source}\n" + f"Message: {payload.message}" + ) + + logger.debug(content) + diff --git a/feature/log_bus_message/log_bus_message.py b/feature/log_bus_message/log_bus_message.py new file mode 100644 index 0000000..7b85254 --- /dev/null +++ b/feature/log_bus_message/log_bus_message.py @@ -0,0 +1,18 @@ +from typing import override + +from kernel.feature.provider import Provider +from kernel.event.registrar import Registrar + +from shared.event.type import Type as EventType + +from feature.log_bus_message.listener.log_message import LogMessage + + +class LogBusMessage(Provider, Registrar): + @override + def register(self) -> None: + pass + + @override + def bootstrap(self) -> None: + self.add_event_listener(EventType.MESSAGE_RECEIVED, LogMessage) diff --git a/kernel/__init__.py b/kernel/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/application/__init__.py b/kernel/application/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/application/application.py b/kernel/application/application.py new file mode 100644 index 0000000..8bfcd1d --- /dev/null +++ b/kernel/application/application.py @@ -0,0 +1,39 @@ +from typing import List, Callable, Concatenate, Awaitable +from abc import ABC, abstractmethod +from punq import Container + +from kernel.feature.provider import Provider +from kernel.application.kernel import Kernel + +class Application(ABC): + __kernel: Kernel + __features: List[Provider] + + @abstractmethod + async def do( + self, + callback: Callable[Concatenate[Container, ...], Awaitable[None]] + ) -> None: + pass + + def __init__(self, kernel: Kernel, features: List[type[Provider]]) -> None: + self.__kernel = kernel + self.__features = [] + + for feature in features: + feature_instance = feature(kernel.get_container()) + feature_instance.register() + + self.__features.append(feature_instance) + + async def launch( + self, + callback: Callable[Concatenate[Container, ...], Awaitable[None]] + ) -> None: + for feature in self.__features: + feature.bootstrap() + + await self.do(callback) + + def get_container(self) -> Container: + return self.__kernel.get_container() diff --git a/kernel/application/enum/__init__.py b/kernel/application/enum/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/application/enum/date_time.py b/kernel/application/enum/date_time.py new file mode 100644 index 0000000..1be0c70 --- /dev/null +++ b/kernel/application/enum/date_time.py @@ -0,0 +1,4 @@ +from enum import Enum + +class DateTime(str, Enum): + Ymd = "%Y-%m-%d" diff --git a/kernel/application/enum/encoding.py b/kernel/application/enum/encoding.py new file mode 100644 index 0000000..937a942 --- /dev/null +++ b/kernel/application/enum/encoding.py @@ -0,0 +1,4 @@ +from enum import Enum + +class Encoding(str, Enum): + utf8 = "utf-8" diff --git a/kernel/application/kernel.py b/kernel/application/kernel.py new file mode 100644 index 0000000..ad5fb27 --- /dev/null +++ b/kernel/application/kernel.py @@ -0,0 +1,5 @@ +from punq import Container + +class Kernel: + def get_container(self) -> Container: + raise NotImplementedError() diff --git a/kernel/application/manager/__init__.py b/kernel/application/manager/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/application/manager/stop_event.py b/kernel/application/manager/stop_event.py new file mode 100644 index 0000000..be42eae --- /dev/null +++ b/kernel/application/manager/stop_event.py @@ -0,0 +1,13 @@ +from asyncio import Event +from signal import signal, SIGINT, SIGTERM + +class StopEvent: + __event: Event + + def __init__(self, event: Event) -> None: + self.__event = event + + def setup_stop_signals(self) -> None: + for signum in (SIGINT, SIGTERM): + signal(signum, lambda sig, frame: self.__event.set()) + diff --git a/kernel/application/stop_event.py b/kernel/application/stop_event.py new file mode 100644 index 0000000..2569e27 --- /dev/null +++ b/kernel/application/stop_event.py @@ -0,0 +1,4 @@ +from asyncio import Event + +class StopEvent(Event): + pass diff --git a/kernel/application/view/__init__.py b/kernel/application/view/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/application/view/application.py b/kernel/application/view/application.py new file mode 100644 index 0000000..97fa286 --- /dev/null +++ b/kernel/application/view/application.py @@ -0,0 +1,7 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Application: + name: str + env: str diff --git a/kernel/application/view/configuration.py b/kernel/application/view/configuration.py new file mode 100644 index 0000000..459ac20 --- /dev/null +++ b/kernel/application/view/configuration.py @@ -0,0 +1,28 @@ +from kernel.application.view.application import Application +from kernel.application.view.rabbitmq import RabbitMQ +from kernel.application.view.redis import Redis + + +class Configuration(): + __application: Application + __rabbitmq: RabbitMQ + __redis: Redis + + def __init__( + self, + application: Application, + rabbitmq: RabbitMQ, + redis: Redis, + ) -> None: + self.__application = application + self.__rabbitmq = rabbitmq + self.__redis = redis + + def application(self) -> Application: + return self.__application + + def rabbitmq(self) -> RabbitMQ: + return self.__rabbitmq + + def redis(self) -> Redis: + return self.__redis diff --git a/kernel/application/view/rabbitmq.py b/kernel/application/view/rabbitmq.py new file mode 100644 index 0000000..86dbdcd --- /dev/null +++ b/kernel/application/view/rabbitmq.py @@ -0,0 +1,11 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class RabbitMQ: + host: str = "localhost" + port: int = 5672 + user: str = "guest" + password: str = "guest" + timeout: int = 10 + vhost: str = "/" diff --git a/kernel/application/view/redis.py b/kernel/application/view/redis.py new file mode 100644 index 0000000..fee41d8 --- /dev/null +++ b/kernel/application/view/redis.py @@ -0,0 +1,8 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Redis: + host: str = "localhost" + port: int = 6379 + db: int = 0 diff --git a/kernel/event/__init__.py b/kernel/event/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/event/listener.py b/kernel/event/listener.py new file mode 100644 index 0000000..3009aee --- /dev/null +++ b/kernel/event/listener.py @@ -0,0 +1,16 @@ +from typing import Generic, TypeVar +from abc import ABC, abstractmethod +from punq import Container + +T = TypeVar('T') + + +class Listener(ABC, Generic[T]): + _container: Container + + def __init__(self, container: Container) -> None: + self._container = container + + @abstractmethod + async def handle(self, event: str, payload: T) -> None: + pass diff --git a/kernel/event/registrar.py b/kernel/event/registrar.py new file mode 100644 index 0000000..6a67859 --- /dev/null +++ b/kernel/event/registrar.py @@ -0,0 +1,25 @@ +from punq import Container +from typing import TypeVar +from pyee import EventEmitter + +from kernel.event.listener import Listener + +T = TypeVar('T') + + +class Registrar: + _container: Container + + def __init__(self, container: Container) -> None: + self._container = container + + def add_event_listener( + self, + event: str, + listener: type[Listener[T]] + ) -> None: + async def callback(payload: T) -> None: + listener_instance = listener(self._container) + await listener_instance.handle(event, payload) + + self._container.resolve(EventEmitter).on(event, callback) diff --git a/kernel/feature/__init__.py b/kernel/feature/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/feature/provider.py b/kernel/feature/provider.py new file mode 100644 index 0000000..4517807 --- /dev/null +++ b/kernel/feature/provider.py @@ -0,0 +1,17 @@ +from abc import ABC, abstractmethod +from punq import Container + + +class Provider(ABC): + _container: Container + + def __init__(self, container: Container) -> None: + self._container = container + + @abstractmethod + def register(self) -> None: + pass + + @abstractmethod + def bootstrap(self) -> None: + pass diff --git a/kernel/rabbitmq/__init__.py b/kernel/rabbitmq/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/rabbitmq/client.py b/kernel/rabbitmq/client.py new file mode 100644 index 0000000..404126c --- /dev/null +++ b/kernel/rabbitmq/client.py @@ -0,0 +1,144 @@ +from asyncio import create_task, Event, sleep +from time import time +from typing import Any, Callable, Union +from logging import Logger, INFO + +from aio_pika import connect_robust, Message, DeliveryMode + +from aio_pika.abc import AbstractIncomingMessage +from aio_pika.abc import AbstractChannel +from aio_pika.abc import AbstractConnection +from aio_pika.abc import AbstractExchange +from aio_pika.abc import AbstractQueue +from aio_pika.exceptions import AMQPConnectionError + +from kernel.rabbitmq.enum.content_type import ContentType +from kernel.rabbitmq.view.authentication_params import AuthenticationParams +from kernel.rabbitmq.view.connection_params import ConnectionParams +from kernel.rabbitmq.exception.connection_timeout import ConnectionTimeout +from kernel.rabbitmq.exception.connection_not_established import ConnectionNotEstablished + + +class Client: + __authentication_params: AuthenticationParams + __connection_params: ConnectionParams + __logger: Union[None, Logger] + __connection: Union[None, AbstractConnection] = None + __stop_event: Union[None, Event] = None + + def __init__( + self, + authentication_params: AuthenticationParams, + connection_params: ConnectionParams, + logger: Union[None, Logger] = None + ) -> None: + self.__authentication_params = authentication_params + self.__connection_params = connection_params + self.__logger = logger + + async def connect(self, stop_event: Event, timeout: int = 15) -> None: + self.__stop_event = stop_event + + ran_at: float = time() + + self.__log(f"Waiting for RabbitMQ connection..") + + while True: + host = self.__connection_params.host + port = self.__connection_params.port + vhost = self.__connection_params.vhost + + try: + self.__connection = await connect_robust( + host=host, + port=port, + virtualhost=vhost, + login=self.__authentication_params.username, + password=self.__authentication_params.password, + ) + + self.__log( + f"Successfully connected to {host}:{port}{vhost}" + ) + + break + except AMQPConnectionError: + seconds_wasted: float = time() - ran_at + + self.__log("Connection failed! Retry after 3s..") + + if seconds_wasted > timeout: + raise ConnectionTimeout( + f"Connection timeout to {host}:{port}{vhost}" + ) + + await sleep(3) + + create_task(self.__wait_for_stop()) + + async def consume( + self, + queue: str, + exchange: str, + consumer: Callable[[AbstractIncomingMessage], Any], + routing_key: Union[None, str] = None, + ) -> None: + self.__raise_if_disconnected() + assert self.__connection is not None + + abstract_channel: AbstractChannel = await self.__connection.channel() + abstract_exchange: AbstractExchange = await abstract_channel\ + .get_exchange(exchange) + abstract_queue: AbstractQueue = await abstract_channel\ + .declare_queue(queue, durable=True) + + await abstract_queue.bind(abstract_exchange, routing_key) + + self.__log( + f"Listening queue {abstract_queue.name} binded to {abstract_exchange.name}..." + ) + + await abstract_queue.consume(consumer) + + async def publish( + self, + exchange: str, + message: str, + routing_key: str = "", + content_type: str = ContentType.APPLICATION_JSON + ) -> None: + self.__raise_if_disconnected() + assert self.__connection is not None + + abstract_exchange: AbstractExchange = await self.__connection.channel()\ + .get_exchange(exchange) + + amqp_message = Message( + body=message.encode(), + content_type=content_type, + delivery_mode=DeliveryMode.PERSISTENT, + ) + + await abstract_exchange.publish(amqp_message, routing_key=routing_key) + + def __raise_if_disconnected(self) -> None: + if self.__connection is None or self.__connection.is_closed: + raise ConnectionNotEstablished("Connection not established") + + async def __wait_for_stop(self) -> None: + if self.__stop_event is not None: + await self.__stop_event.wait() + + self.__log( + "Received stop signal. Terminating RabbitMQ client.." + ) + + connection_established = self.__connection is not None \ + and not self.__connection.is_closed + + if connection_established: + await self.__connection.close() # type: ignore + + def __log(self, message: str, level: int = INFO) -> None: + if self.__logger is not None: + self.__logger.log(level, message) diff --git a/kernel/rabbitmq/enum/__init__.py b/kernel/rabbitmq/enum/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/rabbitmq/enum/content_type.py b/kernel/rabbitmq/enum/content_type.py new file mode 100644 index 0000000..be91987 --- /dev/null +++ b/kernel/rabbitmq/enum/content_type.py @@ -0,0 +1,5 @@ +from enum import Enum + + +class ContentType(str, Enum): + APPLICATION_JSON = "application/json" diff --git a/kernel/rabbitmq/exception/__init__.py b/kernel/rabbitmq/exception/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/rabbitmq/exception/connection_not_established.py b/kernel/rabbitmq/exception/connection_not_established.py new file mode 100644 index 0000000..6fdbbce --- /dev/null +++ b/kernel/rabbitmq/exception/connection_not_established.py @@ -0,0 +1,7 @@ +from aio_pika.exceptions import AMQPConnectionError + +from kernel.rabbitmq.exception.rabbitmq_exception import RabbitMQException + + +class ConnectionNotEstablished(RabbitMQException, AMQPConnectionError): + pass diff --git a/kernel/rabbitmq/exception/connection_timeout.py b/kernel/rabbitmq/exception/connection_timeout.py new file mode 100644 index 0000000..250c807 --- /dev/null +++ b/kernel/rabbitmq/exception/connection_timeout.py @@ -0,0 +1,7 @@ +from aio_pika.exceptions import AMQPConnectionError + +from kernel.rabbitmq.exception.rabbitmq_exception import RabbitMQException + + +class ConnectionTimeout(RabbitMQException, AMQPConnectionError): + pass diff --git a/kernel/rabbitmq/exception/rabbitmq_exception.py b/kernel/rabbitmq/exception/rabbitmq_exception.py new file mode 100644 index 0000000..5a0f990 --- /dev/null +++ b/kernel/rabbitmq/exception/rabbitmq_exception.py @@ -0,0 +1,5 @@ +from aio_pika.exceptions import AMQPException + + +class RabbitMQException(AMQPException): + pass diff --git a/kernel/rabbitmq/view/__init__.py b/kernel/rabbitmq/view/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kernel/rabbitmq/view/authentication_params.py b/kernel/rabbitmq/view/authentication_params.py new file mode 100644 index 0000000..bbdef5a --- /dev/null +++ b/kernel/rabbitmq/view/authentication_params.py @@ -0,0 +1,7 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class AuthenticationParams: + username: str = 'guest' + password: str = 'guest' diff --git a/kernel/rabbitmq/view/connection_params.py b/kernel/rabbitmq/view/connection_params.py new file mode 100644 index 0000000..45d55ea --- /dev/null +++ b/kernel/rabbitmq/view/connection_params.py @@ -0,0 +1,8 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class ConnectionParams: + host: str = "localhost" + port: int = 5672 + vhost: str = "/" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..0ec8285 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,9 @@ +aio-pika==9.5.7 +aioredis==1.3.1 +aioredlock==0.7.3 +pyee==13.0.1 +python-dotenv==1.1.1 +punq==0.7.0 +redis==7.4.0 +setuptools==82.0.1 +xxhash==3.6.0 diff --git a/shared/__init__.py b/shared/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/application/__init__.py b/shared/application/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/application/factory/__init__.py b/shared/application/factory/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/application/factory/application.py b/shared/application/factory/application.py new file mode 100644 index 0000000..6194e58 --- /dev/null +++ b/shared/application/factory/application.py @@ -0,0 +1,10 @@ +from typing import List + +from kernel.feature.provider import Provider + +from shared.application.rabbitmq_application import RabbitMQApplication +from shared.application.rabbitmq_kernel import RabbitMQKernel + +class Application: + def create_rabbitmq(self, features: List[type[Provider]]) -> RabbitMQApplication: + return RabbitMQApplication(RabbitMQKernel(), features) diff --git a/shared/application/factory/logger.py b/shared/application/factory/logger.py new file mode 100644 index 0000000..1dc6fbb --- /dev/null +++ b/shared/application/factory/logger.py @@ -0,0 +1,27 @@ +import logging + +from sys import stdout + + +class Logger: + def create_stdout( + self, + name: str, + level: int = logging.INFO + ) -> logging.Logger: + logger = logging.getLogger(name) + logger.setLevel(level) + + handler = logging.StreamHandler(stdout) + handler.setLevel(level) + handler.setFormatter( + logging.Formatter( + "%(asctime)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S" + ) + ) + + logger.addHandler(handler) + + return logger + diff --git a/shared/application/manager/__init__.py b/shared/application/manager/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/application/manager/listener_arguments_parser.py b/shared/application/manager/listener_arguments_parser.py new file mode 100644 index 0000000..80dad68 --- /dev/null +++ b/shared/application/manager/listener_arguments_parser.py @@ -0,0 +1,40 @@ +from typing import Any, Sequence +from argparse import ArgumentParser, HelpFormatter + + +class ListenerArgumentsParser(ArgumentParser): + def __init__( + self, + prog: str | None = None, + usage: str | None = None, + description: str | None = None, + epilog: str | None = None, + parents: Sequence[ArgumentParser] = [], + formatter_class: type[HelpFormatter] = HelpFormatter, + prefix_chars: str = '-', + fromfile_prefix_chars: str | None = None, + argument_default: Any = None, + conflict_handler: str = 'error', + add_help: bool = True, + allow_abbrev: bool = True, + exit_on_error: bool = True + ) -> None: + super().__init__( + prog=prog, + usage=usage, + description=description, + epilog=epilog, + parents=parents, + formatter_class=formatter_class, + prefix_chars=prefix_chars, + fromfile_prefix_chars=fromfile_prefix_chars, + argument_default=argument_default, + conflict_handler=conflict_handler, + add_help=add_help, + allow_abbrev=allow_abbrev, + exit_on_error=exit_on_error + ) + + self.add_argument('--queue', type=str, required=True) + self.add_argument('--exchange', type=str, required=True) + self.add_argument('--routing-key', type=str, default="") diff --git a/shared/application/rabbitmq_application.py b/shared/application/rabbitmq_application.py new file mode 100644 index 0000000..b415b0a --- /dev/null +++ b/shared/application/rabbitmq_application.py @@ -0,0 +1,56 @@ +from typing import Callable, Awaitable, override +from asyncio import CancelledError +from logging import Logger +from punq import Container +from aio_pika.abc import AbstractIncomingMessage + +from kernel.application.stop_event import StopEvent +from kernel.application.view.configuration import Configuration +from kernel.application.application import Application +from kernel.rabbitmq.client import Client + +from shared.application.view.listener_arguments import ListenerArguments + + +class RabbitMQApplication(Application): + @override + async def do( + self, + callback: Callable[[Container, AbstractIncomingMessage], Awaitable[None]] + ) -> None: + rabbitmq = await self.__connect_rabbitmq_client() + arguments = self.get_container().resolve(ListenerArguments) + + async def consumer(message: AbstractIncomingMessage) -> None: + await callback(self.get_container(), message) + + queue = arguments.queue + exchange = arguments.exchange + routing_key = arguments.routing_key + + await rabbitmq.consume(queue, exchange, consumer, routing_key) + + logger = self.get_container().resolve(Logger) + + logger.info(f"Queue: {queue}") + logger.info(f"Exchange: `{exchange}`") + + if routing_key != "": + logger.info(f"Routing key: {routing_key}") + + try: + await self.get_container().resolve(StopEvent).wait() + except CancelledError as e: + logger.exception(e) + + async def __connect_rabbitmq_client(self) -> Client: + rabbitmq_client = self.get_container().resolve(Client) + + stop_event = self.get_container().resolve(StopEvent) + configuration = self.get_container().resolve(Configuration) + + await rabbitmq_client\ + .connect(stop_event, configuration.rabbitmq().timeout) + + return rabbitmq_client + diff --git a/shared/application/rabbitmq_kernel.py b/shared/application/rabbitmq_kernel.py new file mode 100644 index 0000000..cbd8441 --- /dev/null +++ b/shared/application/rabbitmq_kernel.py @@ -0,0 +1,139 @@ +from os import getenv +from logging import Logger +from punq import Container +from pyee import EventEmitter +from pyee.asyncio import AsyncIOEventEmitter +from aioredlock import Aioredlock +from redis.asyncio import Redis as AsyncRedis + +from kernel.application.manager.stop_event import StopEvent as StopEventManager +from kernel.application.stop_event import StopEvent +from kernel.application.view.application import Application +from kernel.application.view.configuration import Configuration +from kernel.application.view.rabbitmq import RabbitMQ +from kernel.application.view.redis import Redis as RedisConfig + +from kernel.application.kernel import Kernel +from shared.application.service.environment_loader import EnvironmentLoader +from shared.application.manager.listener_arguments_parser import ListenerArgumentsParser +from shared.application.factory.logger import Logger as LoggerFactory +from shared.application.view.listener_arguments import ListenerArguments +from kernel.rabbitmq.client import Client +from kernel.rabbitmq.view.authentication_params import AuthenticationParams +from kernel.rabbitmq.view.connection_params import ConnectionParams + + +class RabbitMQKernel(Kernel): + __container: Container + + def __init__(self) -> None: + self.__load_env() + self.__init_container() + self.__register_config() + self.__register_arguments() + self.__register_stop_event() + self.__register_logger() + self.__register_rabbitmq_client() + self.__register_redis() + self.__register_lock_provider() + self.__register_event_emitter() + + def get_container(self) -> Container: + return self.__container + + def __load_env(self) -> None: + env_loader = EnvironmentLoader(override=False) + env_loader.load() + + def __init_container(self) -> None: + self.__container = Container() + + def __register_config(self) -> None: + application = Application( + name=getenv("APP_NAME", ""), + env=getenv("APP_ENV", ""), + ) + + rabbitmq = RabbitMQ( + host=getenv("RABBITMQ_HOST", "localhost"), + port=int(getenv("RABBITMQ_PORT", "5672")), + user=getenv("RABBITMQ_USER", "guest"), + password=getenv("RABBITMQ_PASS", "guest"), + vhost=getenv("RABBITMQ_VHOST", "/") + ) + + redis = RedisConfig( + host=getenv("REDIS_HOST", "localhost"), + port=int(getenv("REDIS_PORT", "6379")), + db=int(getenv("REDIS_DB", "0")) + ) + + config = Configuration(application, rabbitmq, redis) + + self.__container.register(Configuration, instance=config) + + def __register_arguments(self) -> None: + argument_parser = ListenerArgumentsParser() + arguments_parsed = argument_parser.parse_args() + + arguments = ListenerArguments( + queue=arguments_parsed.queue, + exchange=arguments_parsed.exchange, + routing_key=arguments_parsed.routing_key + ) + + self.__container.register(ListenerArguments, instance=arguments) + + def __register_stop_event(self) -> None: + stop_event = StopEvent() + + stop_event_manager = StopEventManager(stop_event) + stop_event_manager.setup_stop_signals() + + self.__container.register(StopEvent, instance=stop_event) + + def __register_logger(self) -> None: + logger_factory = LoggerFactory() + logger = logger_factory.create_stdout(name="rbmq.listener") + + self.__container.register(Logger, instance=logger) + + def __register_rabbitmq_client(self) -> None: + config = self.__container.resolve(Configuration).rabbitmq() + + rabbitmq_client = Client( + AuthenticationParams(config.user, config.password), + ConnectionParams(config.host, config.port, config.vhost), + self.__container.resolve(Logger) + ) + + self.__container.register(Client, instance=rabbitmq_client) + + def __register_redis(self) -> None: + config = self.__container.resolve(Configuration).redis() + + client = AsyncRedis( + host=config.host, + port=config.port, + db=config.db, + decode_responses=True + ) + + self.__container.register(AsyncRedis, instance=client) + + def __register_lock_provider(self) -> None: + config = self.__container.resolve(Configuration).redis() + + aioredlock = Aioredlock([ + { + "host": config.host, + "port": config.port, + "db": config.db + } + ]) + + self.__container.register(Aioredlock, instance=aioredlock) + + def __register_event_emitter(self) -> None: + event_emitter = AsyncIOEventEmitter() + self.__container.register(EventEmitter, instance=event_emitter) diff --git a/shared/application/service/__init__.py b/shared/application/service/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/application/service/environment_loader.py b/shared/application/service/environment_loader.py new file mode 100644 index 0000000..0704b16 --- /dev/null +++ b/shared/application/service/environment_loader.py @@ -0,0 +1,29 @@ +from dotenv import find_dotenv, load_dotenv + + +class EnvironmentLoader: + __name: str + __usecwd: bool + __override: bool + + def __init__( + self, + name: str = ".env", + usecwd: bool = True, + override: bool = True + ) -> None: + self.__name = name + self.__usecwd = usecwd + self.__override = override + + def load(self) -> None: + dotenv = find_dotenv( + filename=self.__name, + raise_error_if_not_found=True, + usecwd=self.__usecwd + ) + + load_dotenv( + dotenv_path=dotenv, + override=self.__override + ) diff --git a/shared/application/view/__init__.py b/shared/application/view/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/application/view/listener_arguments.py b/shared/application/view/listener_arguments.py new file mode 100644 index 0000000..7fb25b0 --- /dev/null +++ b/shared/application/view/listener_arguments.py @@ -0,0 +1,8 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class ListenerArguments: + queue: str + exchange: str + routing_key: str diff --git a/shared/bus/__init__.py b/shared/bus/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/bus/view/__init__.py b/shared/bus/view/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/bus/view/message.py b/shared/bus/view/message.py new file mode 100644 index 0000000..eae116a --- /dev/null +++ b/shared/bus/view/message.py @@ -0,0 +1,9 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Message: + id: str + type: str + source: str + message: str diff --git a/shared/event/__init__.py b/shared/event/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/event/type.py b/shared/event/type.py new file mode 100644 index 0000000..67d964f --- /dev/null +++ b/shared/event/type.py @@ -0,0 +1,5 @@ +from enum import Enum + + +class Type(str, Enum): + MESSAGE_RECEIVED = "message.received" diff --git a/typings/aioredis/__init__.pyi b/typings/aioredis/__init__.pyi new file mode 100644 index 0000000..a31315c --- /dev/null +++ b/typings/aioredis/__init__.pyi @@ -0,0 +1,321 @@ +from typing import Any, overload, Coroutine, Awaitable + + +class RedisError(Exception): ... + + +class Redis: + """High-level Redis client with all Redis commands.""" + + encoding: str | None + + @overload + def __init__(self, pool_or_conn: Any) -> None: ... + + @overload + def __init__( + self, + pool_or_conn: Any = ..., + *, + host: str, + port: int, + db: int = ..., + password: str | None = ..., + encoding: str | None = ..., + ssl: Any = ..., + ) -> None: ... + + # ==================== Connection Commands ==================== + def close(self) -> None: ... + async def wait_closed(self) -> None: ... + async def auth(self, password: str) -> bool: ... + async def select(self, db: int) -> bool: ... + async def quit(self) -> bool: ... + async def swapdb(self, db1: int, db2: int) -> bool: ... + + # ==================== String Commands ==================== + async def append(self, key: str | bytes, value: str | bytes) -> int: ... + async def bitcount(self, key: str | bytes, start: int | None = ..., end: int | None = ...) -> int: ... + async def bitop_and(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ... + async def bitop_or(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ... + async def bitop_xor(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ... + async def bitop_not(self, destkey: str | bytes, key: str | bytes) -> int: ... + async def bitpos(self, key: str | bytes, bit: int, start: int | None = ..., end: int | None = ...) -> int: ... + async def decr(self, key: str | bytes) -> int: ... + async def decrby(self, key: str | bytes, decrement: int) -> int: ... + async def get(self, key: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ... + async def getbit(self, key: str | bytes, offset: int) -> int: ... + async def getrange(self, key: str | bytes, start: int, end: int, *, encoding: str | None = ...) -> bytes | str: ... + async def getset(self, key: str | bytes, value: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ... + async def incr(self, key: str | bytes) -> int: ... + async def incrby(self, key: str | bytes, increment: int) -> int: ... + async def incrbyfloat(self, key: str | bytes, increment: float) -> float: ... + async def mget(self, key: str | bytes, *keys: str | bytes, encoding: str | None = ...) -> list[Any]: ... + async def mset(self, *args: Any) -> bool: ... + async def msetnx(self, key: str | bytes, value: str | bytes, *pairs: Any) -> bool: ... + async def psetex(self, key: str | bytes, milliseconds: int, value: str | bytes) -> bool: ... + async def set(self, key: str | bytes, value: str | bytes, *, expire: int = ..., pexpire: int = ..., exist: str | None = ...) -> bool: ... + async def setbit(self, key: str | bytes, offset: int, value: int) -> int: ... + async def setex(self, key: str | bytes, seconds: int, value: str | bytes) -> bool: ... + async def setnx(self, key: str | bytes, value: str | bytes) -> bool: ... + async def setrange(self, key: str | bytes, offset: int, value: str | bytes) -> int: ... + async def strlen(self, key: str | bytes) -> int: ... + + # ==================== List Commands ==================== + async def blpop(self, key: str | bytes, *keys: str | bytes, timeout: int = ..., encoding: str | None = ...) -> list[Any] | None: ... + async def brpop(self, key: str | bytes, *keys: str | bytes, timeout: int = ..., encoding: str | None = ...) -> list[Any] | None: ... + async def brpoplpush(self, sourcekey: str | bytes, destkey: str | bytes, timeout: int = ..., encoding: str | None = ...) -> bytes | str | None: ... + async def lindex(self, key: str | bytes, index: int, *, encoding: str | None = ...) -> bytes | str | None: ... + async def linsert(self, key: str | bytes, pivot: str | bytes, value: str | bytes, before: bool = ...) -> int: ... + async def llen(self, key: str | bytes) -> int: ... + async def lpop(self, key: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ... + async def lpush(self, key: str | bytes, value: str | bytes, *values: str | bytes) -> int: ... + async def lpushx(self, key: str | bytes, value: str | bytes) -> int: ... + async def lrange(self, key: str | bytes, start: int, stop: int, *, encoding: str | None = ...) -> list[Any]: ... + async def lrem(self, key: str | bytes, count: int, value: str | bytes) -> int: ... + async def lset(self, key: str | bytes, index: int, value: str | bytes) -> bool: ... + async def ltrim(self, key: str | bytes, start: int, stop: int) -> bool: ... + async def rpop(self, key: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ... + async def rpoplpush(self, sourcekey: str | bytes, destkey: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ... + async def rpush(self, key: str | bytes, value: str | bytes, *values: str | bytes) -> int: ... + async def rpushx(self, key: str | bytes, value: str | bytes) -> int: ... + + # ==================== Set Commands ==================== + async def sadd(self, key: str | bytes, member: str | bytes, *members: str | bytes) -> int: ... + async def scard(self, key: str | bytes) -> int: ... + async def sdiff(self, key: str | bytes, *keys: str | bytes, encoding: str | None = ...) -> list[Any]: ... + async def sdiffstore(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ... + async def sinter(self, key: str | bytes, *keys: str | bytes, encoding: str | None = ...) -> list[Any]: ... + async def sinterstore(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ... + async def sismember(self, key: str | bytes, member: str | bytes) -> bool: ... + async def smembers(self, key: str | bytes, *, encoding: str | None = ...) -> list[Any]: ... + async def smove(self, sourcekey: str | bytes, destkey: str | bytes, member: str | bytes) -> bool: ... + async def spop(self, key: str | bytes, count: int | None = ..., *, encoding: str | None = ...) -> bytes | str | list[Any] | None: ... + async def srandmember(self, key: str | bytes, count: int | None = ..., *, encoding: str | None = ...) -> bytes | str | list[Any] | None: ... + async def srem(self, key: str | bytes, member: str | bytes, *members: str | bytes) -> int: ... + async def sscan(self, key: str | bytes, cursor: int = ..., match: str | None = ..., count: int | None = ...) -> tuple[int, list[Any]]: ... + + # ==================== Hash Commands ==================== + async def hdel(self, key: str | bytes, field: str | bytes, *fields: str | bytes) -> int: ... + async def hexists(self, key: str | bytes, field: str | bytes) -> bool: ... + async def hget(self, key: str | bytes, field: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ... + async def hgetall(self, key: str | bytes, *, encoding: str | None = ...) -> dict[Any, Any]: ... + async def hincrby(self, key: str | bytes, field: str | bytes, increment: int) -> int: ... + async def hincrbyfloat(self, key: str | bytes, field: str | bytes, increment: float) -> float: ... + async def hkeys(self, key: str | bytes, *, encoding: str | None = ...) -> list[Any]: ... + async def hlen(self, key: str | bytes) -> int: ... + async def hmget(self, key: str | bytes, field: str | bytes, *fields: str | bytes, encoding: str | None = ...) -> list[Any]: ... + async def hmset(self, key: str | bytes, field: str | bytes, value: str | bytes, *fv_pairs: Any) -> bool: ... + async def hmset_dict(self, key: str | bytes, mapping: dict[Any, Any]) -> bool: ... + async def hset(self, key: str | bytes, field: str | bytes, value: str | bytes) -> int: ... + async def hsetnx(self, key: str | bytes, field: str | bytes, value: str | bytes) -> bool: ... + async def hstrlen(self, key: str | bytes, field: str | bytes) -> int: ... + async def hvals(self, key: str | bytes, *, encoding: str | None = ...) -> list[Any]: ... + async def hscan(self, key: str | bytes, cursor: int = ..., match: str | None = ..., count: int | None = ...) -> tuple[int, list[Any]]: ... + + # ==================== Sorted Set Commands ==================== + async def zadd(self, key: str | bytes, score: float, member: str | bytes, *score_members: Any) -> int: ... + async def zcard(self, key: str | bytes) -> int: ... + async def zcount(self, key: str | bytes, min: float | str, max: float | str) -> int: ... + async def zincrby(self, key: str | bytes, increment: float, member: str | bytes) -> float: ... + async def zinterstore(self, destkey: str | bytes, numkeys: int, key: str | bytes, *keys: str | bytes, weights: list[int] | None = ..., aggregate: str | None = ...) -> int: ... + async def zlexcount(self, key: str | bytes, min: str, max: str) -> int: ... + async def zpopmax(self, key: str | bytes, count: int | None = ..., encoding: str | None = ...) -> list[Any]: ... + async def zpopmin(self, key: str | bytes, count: int | None = ..., encoding: str | None = ...) -> list[Any]: ... + async def zrange(self, key: str | bytes, start: int, stop: int, *, withscores: bool = ..., encoding: str | None = ...) -> list[Any]: ... + async def zrangebylex(self, key: str | bytes, min: str, max: str, *, offset: int | None = ..., count: int | None = ..., encoding: str | None = ...) -> list[Any]: ... + async def zrangebyscore(self, key: str | bytes, min: float | str, max: float | str, *, offset: int | None = ..., count: int | None = ..., withscores: bool = ..., encoding: str | None = ...) -> list[Any]: ... + async def zrank(self, key: str | bytes, member: str | bytes) -> int | None: ... + async def zrem(self, key: str | bytes, member: str | bytes, *members: str | bytes) -> int: ... + async def zremrangebylex(self, key: str | bytes, min: str, max: str) -> int: ... + async def zremrangebyrank(self, key: str | bytes, start: int, stop: int) -> int: ... + async def zremrangebyscore(self, key: str | bytes, min: float | str, max: float | str) -> int: ... + async def zrevrange(self, key: str | bytes, start: int, stop: int, *, withscores: bool = ..., encoding: str | None = ...) -> list[Any]: ... + async def zrevrangebylex(self, key: str | bytes, max: str, min: str, *, offset: int | None = ..., count: int | None = ..., encoding: str | None = ...) -> list[Any]: ... + async def zrevrangebyscore(self, key: str | bytes, max: float | str, min: float | str, *, offset: int | None = ..., count: int | None = ..., withscores: bool = ..., encoding: str | None = ...) -> list[Any]: ... + async def zrevrank(self, key: str | bytes, member: str | bytes) -> int | None: ... + async def zscan(self, key: str | bytes, cursor: int = ..., match: str | None = ..., count: int | None = ...) -> tuple[int, list[Any]]: ... + async def zscore(self, key: str | bytes, member: str | bytes) -> float | None: ... + async def zunionstore(self, destkey: str | bytes, numkeys: int, key: str | bytes, *keys: str | bytes, weights: list[int] | None = ..., aggregate: str | None = ...) -> int: ... + async def bzpopmax(self, key: str | bytes, *keys: str | bytes, timeout: int = ..., encoding: str | None = ...) -> list[Any] | None: ... + async def bzpopmin(self, key: str | bytes, *keys: str | bytes, timeout: int = ..., encoding: str | None = ...) -> list[Any] | None: ... + + # ==================== Generic/Key Commands ==================== + async def delete(self, key: str | bytes, *keys: str | bytes) -> int: ... + async def dump(self, key: str | bytes) -> bytes | None: ... + async def exists(self, key: str | bytes, *keys: str | bytes) -> int: ... + async def expire(self, key: str | bytes, timeout: int) -> bool: ... + async def expireat(self, key: str | bytes, timestamp: int) -> bool: ... + async def keys(self, pattern: str | bytes, *, encoding: str | None = ...) -> list[Any]: ... + async def migrate(self, host: str, port: int, key: str | bytes, dest_db: int, timeout: int, *, copy: bool = ..., replace: bool = ...) -> bool: ... + async def move(self, key: str | bytes, db: int) -> bool: ... + async def object_encoding(self, key: str | bytes) -> bytes | str: ... + async def object_idletime(self, key: str | bytes) -> int | None: ... + async def object_refcount(self, key: str | bytes) -> int | None: ... + async def persist(self, key: str | bytes) -> bool: ... + async def pexpire(self, key: str | bytes, timeout: int) -> bool: ... + async def pexpireat(self, key: str | bytes, timestamp: int) -> bool: ... + async def pttl(self, key: str | bytes) -> int: ... + async def randomkey(self, *, encoding: str | None = ...) -> bytes | str | None: ... + async def rename(self, key: str | bytes, newkey: str | bytes) -> bool: ... + async def renamenx(self, key: str | bytes, newkey: str | bytes) -> bool: ... + async def restore(self, key: str | bytes, ttl: int, value: bytes) -> bool: ... + async def scan(self, cursor: int = ..., match: str | None = ..., count: int | None = ...) -> tuple[int, list[Any]]: ... + async def sort(self, key: str | bytes, *get_patterns: str | bytes, by: str | None = ..., offset: int | None = ..., count: int | None = ..., asc: bool | None = ..., alpha: bool = ..., store: str | bytes | None = ...) -> list[Any]: ... + async def touch(self, key: str | bytes, *keys: str | bytes) -> int: ... + async def ttl(self, key: str | bytes) -> int: ... + async def type(self, key: str | bytes) -> bytes: ... + async def unlink(self, key: str | bytes, *keys: str | bytes) -> int: ... + async def wait(self, numslaves: int, timeout: int) -> int: ... + + # ==================== Pub/Sub Commands ==================== + async def publish(self, channel: str | bytes, message: str | bytes) -> int: ... + async def publish_json(self, channel: str | bytes, obj: Any) -> int: ... + async def pubsub_channels(self, pattern: str | None = ..., *, encoding: str | None = ...) -> list[Any]: ... + async def pubsub_numpat(self) -> int: ... + async def pubsub_numsub(self, *channels: str | bytes) -> dict[Any, int]: ... + async def subscribe(self, channel: str | bytes, *channels: str | bytes) -> list[Any]: ... + async def unsubscribe(self, channel: str | bytes | None = ..., *channels: str | bytes) -> list[Any]: ... + async def psubscribe(self, pattern: str | bytes, *patterns: str | bytes) -> list[Any]: ... + async def punsubscribe(self, pattern: str | bytes | None = ..., *patterns: str | bytes) -> list[Any]: ... + + # ==================== Transaction Commands ==================== + async def multi_exec(self, *commands: Any) -> list[Any]: ... + async def watch(self, key: str | bytes, *keys: str | bytes) -> bool: ... + async def unwatch(self) -> bool: ... + + # ==================== Scripting Commands ==================== + async def eval(self, script: str | bytes, keys: list[str | bytes] | None = ..., args: list[Any] | None = ...) -> Any: ... + async def evalsha(self, digest: str | bytes, keys: list[str | bytes] | None = ..., args: list[Any] | None = ...) -> Any: ... + async def script_exists(self, digest: str | bytes, *digests: str | bytes) -> list[bool]: ... + async def script_flush(self) -> bool: ... + async def script_kill(self) -> bool: ... + async def script_load(self, script: str | bytes) -> bytes: ... + + # ==================== Server Commands ==================== + async def bgrewriteaof(self) -> bytes: ... + async def bgsave(self) -> bytes: ... + async def client_getname(self, *, encoding: str | None = ...) -> bytes | str | None: ... + async def client_kill(self, *args: Any) -> int: ... + async def client_list(self, *, encoding: str | None = ...) -> bytes | str: ... + async def client_pause(self, timeout: int) -> bool: ... + async def client_reply(self, *args: Any) -> bool: ... + async def client_setname(self, name: str | bytes) -> bool: ... + async def command(self) -> list[Any]: ... + async def command_count(self) -> int: ... + async def command_getkeys(self, command: str | bytes, *args: Any, encoding: str = ...) -> list[Any]: ... + async def command_info(self, command: str | bytes, *commands: str | bytes) -> list[Any]: ... + async def config_get(self, parameter: str | bytes = ..., *, encoding: str | None = ...) -> dict[Any, Any]: ... + async def config_resetstat(self) -> bool: ... + async def config_rewrite(self) -> bool: ... + async def config_set(self, parameter: str | bytes, value: str | bytes) -> bool: ... + async def dbsize(self) -> int: ... + async def debug_object(self, key: str | bytes) -> bytes: ... + async def debug_sleep(self, timeout: float) -> None: ... + async def flushall(self, async_mode: bool = ...) -> bool: ... + async def flushdb(self, async_mode: bool = ...) -> bool: ... + async def info(self, section: str | None = ..., *, encoding: str | None = ...) -> bytes | str | dict[str, Any]: ... + async def lastsave(self) -> int: ... + async def monitor(self) -> Any: ... + async def ping(self, message: str | bytes | None = ..., *, encoding: str | None = ...) -> bytes | str: ... + async def save(self) -> bool: ... + async def shutdown(self, save: bool = ...) -> None: ... + async def slaveof(self, host: str, port: int) -> bool: ... + async def slowlog_get(self, count: int | None = ...) -> list[Any]: ... + async def slowlog_len(self) -> int: ... + async def slowlog_reset(self) -> bool: ... + async def sync(self) -> Any: ... + async def time(self) -> tuple[int, int]: ... + async def role(self) -> list[Any]: ... + + # ==================== Geo Commands ==================== + async def geoadd(self, key: str | bytes, longitude: float, latitude: float, member: str | bytes, *latlng_member: Any) -> int: ... + async def geodist(self, key: str | bytes, member1: str | bytes, member2: str | bytes, unit: str | None = ...) -> float | None: ... + async def geohash(self, key: str | bytes, member: str | bytes, *members: str | bytes, encoding: str | None = ...) -> list[Any]: ... + async def geopos(self, key: str | bytes, member: str | bytes, *members: str | bytes) -> list[Any]: ... + async def georadius(self, key: str | bytes, longitude: float, latitude: float, radius: float, unit: str, *, count: int | None = ..., sort: str | None = ..., encoding: str | None = ...) -> list[Any]: ... + async def georadiusbymember(self, key: str | bytes, member: str | bytes, radius: float, unit: str, *, count: int | None = ..., sort: str | None = ..., encoding: str | None = ...) -> list[Any]: ... + + # ==================== HyperLogLog Commands ==================== + async def pfadd(self, key: str | bytes, element: str | bytes, *elements: str | bytes) -> int: ... + async def pfcount(self, key: str | bytes, *keys: str | bytes) -> int: ... + async def pfmerge(self, destkey: str | bytes, sourcekey: str | bytes, *sourcekeys: str | bytes) -> bool: ... + + # ==================== Stream Commands ==================== + async def xack(self, key: str | bytes, group: str | bytes, id: str | bytes, *ids: str | bytes) -> int: ... + async def xadd(self, key: str | bytes, id: str | bytes, *field_value_pairs: Any) -> bytes: ... + async def xclaim(self, key: str | bytes, group: str | bytes, consumer: str | bytes, min_idle_time: int, id: str | bytes, *ids: str | bytes) -> list[Any]: ... + async def xdel(self, key: str | bytes, id: str | bytes, *ids: str | bytes) -> int: ... + async def xgroup_create(self, key: str | bytes, group: str | bytes, id: str | bytes = ...) -> bool: ... + async def xgroup_delconsumer(self, key: str | bytes, group: str | bytes, consumer: str | bytes) -> int: ... + async def xgroup_destroy(self, key: str | bytes, group: str | bytes) -> int: ... + async def xgroup_setid(self, key: str | bytes, group: str | bytes, id: str | bytes) -> bool: ... + async def xlen(self, key: str | bytes) -> int: ... + async def xpending(self, key: str | bytes, group: str | bytes) -> list[Any]: ... + async def xrange(self, key: str | bytes, start: str | bytes, end: str | bytes, count: int | None = ...) -> list[Any]: ... + async def xread(self, *keys_ids: Any, count: int | None = ..., timeout: int | None = ...) -> list[Any]: ... + async def xread_group(self, group: str | bytes, consumer: str | bytes, *keys_ids: Any, count: int | None = ..., timeout: int | None = ...) -> list[Any]: ... + async def xrevrange(self, key: str | bytes, end: str | bytes, start: str | bytes, count: int | None = ...) -> list[Any]: ... + async def xtrim(self, key: str | bytes, maxlen: int, approximate: bool = ...) -> int: ... + + # ==================== Connection Pool Properties ==================== + @property + def db(self) -> int: ... + @property + def closed(self) -> bool: ... + @property + def address(self) -> tuple[str, int] | str: ... + @property + def connection(self) -> Any: ... + @property + def in_transaction(self) -> bool: ... + + +async def create_redis( + address: str | tuple[str, int], + *, + db: int = ..., + password: str | None = ..., + encoding: str | None = ..., + ssl: Any = ..., + minsize: int = ..., + maxsize: int = ..., + parser: Any = ..., + create_connection_timeout: float | None = ..., +) -> Redis: ... + + +async def create_redis_pool( + address: str | tuple[str, int], + *, + db: int = ..., + password: str | None = ..., + encoding: str | None = ..., + ssl: Any = ..., + minsize: int = ..., + maxsize: int = ..., + parser: Any = ..., + create_connection_timeout: float | None = ..., +) -> Redis: ... + + +# Errors and Exceptions +class ProtocolError(RedisError): ... +class ReplyError(RedisError): ... +class ConnectionClosedError(RedisError): ... +class ConnectionForcedCloseError(RedisError): ... +class MasterNotFoundError(RedisError): ... +class MultiExecError(RedisError): ... +class PipelineError(RedisError): ... +class ReadOnlyError(RedisError): ... +class MaxClientsError(RedisError): ... +class AuthError(RedisError): ... +class ChannelClosedError(RedisError): ... +class WatchVariableError(RedisError): ... +class PoolClosedError(RedisError): ... +class SlaveNotFoundError(RedisError): ... +class MasterReplyError(RedisError): ... +class SlaveReplyError(RedisError): ... + + +__all__: tuple[str, ...] diff --git a/typings/aioredlock/__init__.pyi b/typings/aioredlock/__init__.pyi new file mode 100644 index 0000000..e8bc950 --- /dev/null +++ b/typings/aioredlock/__init__.pyi @@ -0,0 +1,95 @@ +from typing import Any, TypeAlias +import ssl + +_RedisAddress: TypeAlias = tuple[str, int] | list[tuple[str, int]] +_RedisConnection: TypeAlias = dict[str, Any] | str | _RedisAddress | Any + + +class AioredlockError(Exception): ... + + +class LockError(AioredlockError): ... + + +class LockAcquiringError(LockError): ... + + +class LockRuntimeError(LockError): ... + + +class SentinelConfigError(Exception): ... + + +class Sentinel: + master: str + connection: list[tuple[str, int]] + redis_kwargs: dict[str, Any] + + def __init__( + self, + connection: dict[str, Any] | str | _RedisAddress, + master: str | None = ..., + password: str | None = ..., + db: int | None = ..., + ssl_context: ssl.SSLContext | bool | None = ..., + ) -> None: ... + + async def get_sentinel(self) -> Any: ... + async def get_master(self) -> Any: ... + + +class Lock: + lock_manager: Aioredlock + resource: str + id: str + lock_timeout: float | None + valid: bool + + def __init__( + self, + lock_manager: Aioredlock, + resource: str, + id: str, + lock_timeout: float | None = ..., + valid: bool = ..., + ) -> None: ... + + async def __aenter__(self) -> Lock: ... + async def __aexit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: Any) -> None: ... + async def extend(self) -> None: ... + async def release(self) -> None: ... + async def is_locked(self) -> bool: ... + + +class Aioredlock: + redis_connections: list[_RedisConnection] + retry_count: int + retry_delay_min: float + retry_delay_max: float + internal_lock_timeout: float + + def __init__( + self, + redis_connections: list[_RedisConnection] = ..., + retry_count: int = ..., + retry_delay_min: float = ..., + retry_delay_max: float = ..., + internal_lock_timeout: float = ..., + ) -> None: ... + + async def lock( + self, + resource: str, + lock_timeout: float | None = ..., + lock_identifier: str | None = ..., + ) -> Lock: ... + + async def extend(self, lock: Lock, lock_timeout: float | None = ...) -> None: ... + async def unlock(self, lock: Lock) -> None: ... + async def is_locked(self, resource_or_lock: str | Lock) -> bool: ... + async def destroy(self) -> None: ... + async def get_active_locks(self) -> list[Lock]: ... + async def get_lock(self, resource: str, lock_identifier: str) -> Lock: ... + + +__all__: tuple[str, ...] diff --git a/typings/punq/__init__.pyi b/typings/punq/__init__.pyi new file mode 100644 index 0000000..0a67754 --- /dev/null +++ b/typings/punq/__init__.pyi @@ -0,0 +1,55 @@ +from enum import Enum +from typing import Any, Callable, List, Type, TypeVar, overload + +T = TypeVar("T") + +class MissingDependencyException(Exception): ... +class MissingDependencyError(MissingDependencyException): ... +class InvalidRegistrationException(Exception): ... +class InvalidRegistrationError(InvalidRegistrationException): ... +class InvalidForwardReferenceException(Exception): ... +class InvalidForwardReferenceError(InvalidForwardReferenceException): ... + +class Scope(Enum): + transient = 0 + singleton = 1 + +class _Empty: ... + +empty: _Empty + +class Container: + def __init__(self) -> None: ... + + @overload + def register( + self, + service: Type[T], + factory: Type[T] | Callable[..., T] | _Empty = ..., + instance: T | _Empty = ..., + scope: Scope = ..., + **kwargs: Any, + ) -> Container: ... + + @overload + def register( + self, + service: str, + factory: Callable[..., Any] | _Empty = ..., + instance: Any | _Empty = ..., + scope: Scope = ..., + **kwargs: Any, + ) -> Container: ... + + def register( + self, + service: Type[T] | str, + factory: Type[T] | Callable[..., T] | _Empty = ..., + instance: T | _Empty = ..., + scope: Scope = ..., + **kwargs: Any, + ) -> Container: ... + + def resolve(self, service_key: Type[T], **kwargs: Any) -> T: ... + def resolve_all(self, service: Type[T], **kwargs: Any) -> List[T]: ... + def instantiate(self, service_key: Type[T], **kwargs: Any) -> T: ...