Thank you for your purchase. Your invoice is attached.
+If you have any questions, please contact support.
+ + + """ + + # Attachments + attachments = [ + { + "file_path": invoice_pdf_path, + "filename": f"Invoice-{invoice_number}.pdf", + "mime_type": "application/pdf" + } + ] + + # Send email with attachment + email_adapter.send( + to=user_email, + subject=f"Your Invoice #{invoice_number}", + body=html_content, + body_type="html", + attachments=attachments + ) + + # Usage + send_invoice_email( + "customer@example.com", + "INV-12345", + "/path/to/invoices/INV-12345.pdf" + ) + +Multiple Recipients and CC/BCC +---------------------------- + +Send to multiple recipients with CC and BCC: + +.. code-block:: python + + from archipy.adapters.email.email_adapter import EmailAdapter + from pydantic import EmailStr + + email_adapter = EmailAdapter() + + def send_team_notification(subject, message, team_emails, cc_manager=True): + # Convert recipient list to expected format + recipients = [EmailStr(email) for email in team_emails] + + # Add CC recipients if needed + cc = [] + if cc_manager: + cc = [EmailStr("manager@example.com")] + + email_adapter.send( + to=recipients, + subject=subject, + body=message, + body_type="plain", + cc=cc, + bcc=[EmailStr("records@example.com")] # BCC for record keeping + ) + + # Usage + send_team_notification( + "Project Update", + "The project milestone has been completed.", + ["team1@example.com", "team2@example.com"] + ) + +Template-Based Emails +------------------ + +Send emails using templates: + +.. code-block:: python + + from archipy.adapters.email.email_adapter import EmailAdapter + import os + + email_adapter = EmailAdapter() + + def send_password_reset(user_email, reset_token, user_name): + # Read template from file + template_path = os.path.join("templates", "emails", "password_reset.html") + with open(template_path, "r") as file: + template = file.read() + + # Replace placeholders in template + html_content = template.replace("{{user_name}}", user_name) + html_content = html_content.replace("{{reset_token}}", reset_token) + html_content = html_content.replace( + "{{reset_link}}", + f"https://example.com/reset-password?token={reset_token}" + ) + + # Send email using template + email_adapter.send( + to=user_email, + subject="Password Reset Request", + body=html_content, + body_type="html" + ) + + # Usage + send_password_reset( + "user@example.com", + "abc123xyz789", + "John Doe" + ) diff --git a/docs/source/examples/adapters/index.rst b/docs/source/examples/adapters/index.rst new file mode 100644 index 00000000..98579d93 --- /dev/null +++ b/docs/source/examples/adapters/index.rst @@ -0,0 +1,25 @@ +.. _examples_adapters: + +Adapter Utilities +=============== + +ArchiPy provides adapters for interfacing with external systems: + +.. toctree:: + :maxdepth: 1 + :caption: Adapter Modules: + + orm + redis + email + +Overview +-------- + +Adapters isolate your core application from external dependencies: + +* **ORM Adapters**: Interfaces for database operations (SQLAlchemy) +* **Redis Adapters**: Key-value store and caching adapters +* **Email Adapters**: Email delivery interfaces + +All adapters follow the ports and adapters pattern, with corresponding interfaces (ports) and implementations (adapters). diff --git a/docs/source/examples/adapters/orm.rst b/docs/source/examples/adapters/orm.rst new file mode 100644 index 00000000..00065c80 --- /dev/null +++ b/docs/source/examples/adapters/orm.rst @@ -0,0 +1,178 @@ +.. _examples_adapters_orm: + +ORM Adapters +========== + +Database operations using SQLAlchemy adapters. + +Configuration +------------ + +First, configure your database settings: + +.. code-block:: python + + from archipy.configs.base_config import BaseConfig + + class AppConfig(BaseConfig): + SQLALCHEMY = { + "DRIVER_NAME": "postgresql", + "USERNAME": "postgres", + "PASSWORD": "password", + "HOST": "localhost", + "PORT": 5432, + "DATABASE": "myapp", + "POOL_SIZE": 10 + } + + # Set global configuration + config = AppConfig() + BaseConfig.set_global(config) + +Defining Models +------------- + +Define your database models: + +.. code-block:: python + + from uuid import uuid4 + from sqlalchemy import Column, String, ForeignKey + from sqlalchemy.orm import relationship + from archipy.models.entities import BaseEntity + + class User(BaseEntity): + __tablename__ = "users" + + name = Column(String(100)) + email = Column(String(100), unique=True) + + # Relationships + posts = relationship("Post", back_populates="author") + + class Post(BaseEntity): + __tablename__ = "posts" + + title = Column(String(255)) + content = Column(String(1000)) + + # Foreign keys + author_id = Column(String(36), ForeignKey("users.test_uuid")) + + # Relationships + author = relationship("User", back_populates="posts") + +Synchronous Operations +------------------- + +Examples of synchronous database operations: + +.. code-block:: python + + from archipy.adapters.orm.sqlalchemy.session_manager_adapters import SessionManagerAdapter + from archipy.adapters.orm.sqlalchemy.sqlalchemy_adapters import SqlAlchemyAdapter + from sqlalchemy import select + + # Create session manager and adapter + session_manager = SessionManagerAdapter() + adapter = SqlAlchemyAdapter(session_manager) + + # Create a user + def create_user(name, email): + user = User(test_uuid=uuid4(), name=name, email=email) + return adapter.create(user) + + # Query users + def find_users_by_name(name): + query = select(User).where(User.name == name) + users, total = adapter.execute_search_query(User, query) + return users + + # Update a user + def update_user_email(user_id, new_email): + user = adapter.get_by_uuid(User, user_id) + if user: + user.email = new_email + return adapter.update(user) + return None + + # Delete a user + def delete_user(user_id): + user = adapter.get_by_uuid(User, user_id) + if user: + adapter.delete(user) + return True + return False + +Asynchronous Operations +-------------------- + +Examples of asynchronous database operations: + +.. code-block:: python + + import asyncio + from archipy.adapters.orm.sqlalchemy.session_manager_adapters import AsyncSessionManagerAdapter + from archipy.adapters.orm.sqlalchemy.sqlalchemy_adapters import AsyncSqlAlchemyAdapter + + # Create async session manager and adapter + async_session_manager = AsyncSessionManagerAdapter() + async_adapter = AsyncSqlAlchemyAdapter(async_session_manager) + + # Create a user asynchronously + async def create_user_async(name, email): + user = User(test_uuid=uuid4(), name=name, email=email) + return await async_adapter.create(user) + + # Query users asynchronously + async def find_users_by_name_async(name): + query = select(User).where(User.name == name) + users, total = await async_adapter.execute_search_query(User, query) + return users + + # Usage with asyncio + async def main(): + # Create a user + user = await create_user_async("John Doe", "john@example.com") + + # Find users + users = await find_users_by_name_async("John Doe") + print(f"Found {len(users)} users") + + # Run the async function + asyncio.run(main()) + +Transactions +----------- + +Using transactions to ensure consistency: + +.. code-block:: python + + from archipy.helpers.utils.atomic_transaction import atomic_transaction, async_atomic_transaction + + # Synchronous transaction + def transfer_points(from_user_id, to_user_id, amount): + with atomic_transaction(adapter.session_manager): + from_user = adapter.get_by_uuid(User, from_user_id) + to_user = adapter.get_by_uuid(User, to_user_id) + + # Perform operations atomically + from_user.points -= amount + to_user.points += amount + + adapter.update(from_user) + adapter.update(to_user) + + # Asynchronous transaction + async def transfer_points_async(from_user_id, to_user_id, amount): + async with async_atomic_transaction(async_adapter.session_manager): + from_user = await async_adapter.get_by_uuid(User, from_user_id) + to_user = await async_adapter.get_by_uuid(User, to_user_id) + + # Perform operations atomically + from_user.points -= amount + to_user.points += amount + + await async_adapter.update(from_user) + await async_adapter.update(to_user) diff --git a/docs/source/examples/adapters/redis.rst b/docs/source/examples/adapters/redis.rst new file mode 100644 index 00000000..64d0f521 --- /dev/null +++ b/docs/source/examples/adapters/redis.rst @@ -0,0 +1,185 @@ +.. _examples_adapters_redis: + +Redis Adapters +============ + +Working with Redis for caching and messaging. + +Configuration +------------ + +First, configure your Redis settings: + +.. code-block:: python + + from archipy.configs.base_config import BaseConfig + + class AppConfig(BaseConfig): + REDIS = { + "MASTER_HOST": "localhost", + "PORT": 6379, + "PASSWORD": "password", + "DB": 0, + "DECODE_RESPONSES": True + } + + # Set global configuration + config = AppConfig() + BaseConfig.set_global(config) + +Synchronous Operations +------------------- + +Examples of synchronous Redis operations: + +.. code-block:: python + + import json + from archipy.adapters.redis.redis_adapters import RedisAdapter + + # Create Redis adapter + redis = RedisAdapter() + + # Basic operations + def cache_user(user_id, user_data): + # Cache user data with 1 hour expiry + redis.set(f"user:{user_id}", json.dumps(user_data), ex=3600) + + def get_cached_user(user_id): + data = redis.get(f"user:{user_id}") + if data: + return json.loads(data) + return None + + def delete_cached_user(user_id): + redis.delete(f"user:{user_id}") + + # List operations + def add_to_recent_users(user_id): + # Add to list and trim to 10 most recent + redis.lpush("recent_users", user_id) + redis.ltrim("recent_users", 0, 9) + + def get_recent_users(): + return redis.lrange("recent_users", 0, -1) + + # Set operations + def add_user_role(user_id, role): + redis.sadd(f"user:{user_id}:roles", role) + + def has_user_role(user_id, role): + return redis.sismember(f"user:{user_id}:roles", role) + + def get_user_roles(user_id): + return redis.smembers(f"user:{user_id}:roles") + +Asynchronous Operations +-------------------- + +Examples of asynchronous Redis operations: + +.. code-block:: python + + import asyncio + import json + from archipy.adapters.redis.redis_adapters import AsyncRedisAdapter + + # Create async Redis adapter + async_redis = AsyncRedisAdapter() + + # Basic async operations + async def cache_user_async(user_id, user_data): + # Cache user data with 1 hour expiry + await async_redis.set(f"user:{user_id}", json.dumps(user_data), ex=3600) + + async def get_cached_user_async(user_id): + data = await async_redis.get(f"user:{user_id}") + if data: + return json.loads(data) + return None + + # Hash operations + async def update_user_profile(user_id, **fields): + # Update multiple fields in a hash + await async_redis.hset(f"profile:{user_id}", mapping=fields) + + async def get_user_profile(user_id): + # Get all fields from a hash + return await async_redis.hgetall(f"profile:{user_id}") + + # Usage with asyncio + async def main(): + # Cache a user + await cache_user_async("123", {"name": "John", "email": "john@example.com"}) + + # Update profile + await update_user_profile("123", + first_name="John", + last_name="Doe", + age=30) + + # Get cached data + user = await get_cached_user_async("123") + profile = await get_user_profile("123") + + print(f"User: {user}") + print(f"Profile: {profile}") + + # Run the async function + asyncio.run(main()) + +Pub/Sub Messaging +--------------- + +Using Redis for publish/subscribe messaging: + +.. code-block:: python + + import asyncio + import json + from archipy.adapters.redis.redis_adapters import AsyncRedisAdapter + + async_redis = AsyncRedisAdapter() + + # Publisher + async def publish_event(event_type, data): + message = json.dumps({ + "type": event_type, + "data": data, + "timestamp": datetime.utcnow().isoformat() + }) + await async_redis.publish(f"events:{event_type}", message) + + # Subscriber + async def subscribe_to_events(): + # Create a new connection for the subscription + pubsub = async_redis.pubsub() + + # Subscribe to channels + await pubsub.subscribe("events:user_created", "events:user_updated") + + # Process messages + async for message in pubsub.listen(): + if message["type"] == "message": + data = json.loads(message["data"]) + print(f"Received event: {data['type']}") + # Process event based on type + if data["type"] == "user_created": + await process_user_created(data["data"]) + elif data["type"] == "user_updated": + await process_user_updated(data["data"]) + + # Usage + async def main(): + # Start subscriber in the background + asyncio.create_task(subscribe_to_events()) + + # Publish events + await publish_event("user_created", {"id": "123", "name": "Alice"}) + await publish_event("user_updated", {"id": "123", "name": "Alice Smith"}) + + # Keep running to receive messages + await asyncio.sleep(10) + + # Run the async function + asyncio.run(main()) diff --git a/docs/source/examples/bdd_testing.rst b/docs/source/examples/bdd_testing.rst new file mode 100644 index 00000000..3c6c7e4d --- /dev/null +++ b/docs/source/examples/bdd_testing.rst @@ -0,0 +1,330 @@ +.. _examples_bdd: + +BDD Testing with ArchiPy +======================= + +ArchiPy provides comprehensive support for Behavior-Driven Development testing using Behave. + +Feature File Example +------------------ + +.. code-block:: gherkin + + # features/user_management.feature + Feature: User Management + + Background: + Given the application database is initialized + + Scenario: Create a new user + When a new user "John Doe" with email "john@example.com" is created + Then the user should exist in the database + And the user should have the correct email "john@example.com" + + @async + Scenario: Retrieve user asynchronously + Given a user with email "jane@example.com" exists + When I retrieve the user by email asynchronously + Then the user details should be retrieved successfully + +Environment Setup +--------------- + +ArchiPy's BDD framework uses scenario context isolation and pooling: + +.. code-block:: python + + # features/environment.py + import logging + import uuid + from behave.model import Scenario + from behave.runner import Context + + from archipy.configs.base_config import BaseConfig + from features.scenario_context_pool_manager import ScenarioContextPoolManager + + # Define test configuration + class TestConfig(BaseConfig): + # Test-specific configurations + SQLALCHEMY = { + "DRIVER_NAME": "sqlite", + "DATABASE": ":memory:", + "ISOLATION_LEVEL": None + } + + def before_all(context: Context): + """Setup performed once at the start of the test run.""" + # Set up logging + context.logger = logging.getLogger("behave") + + # Create scenario context pool + context.scenario_context_pool = ScenarioContextPoolManager() + + # Set global configuration for tests + config = TestConfig() + BaseConfig.set_global(config) + + context.logger.info("Test environment initialized") + + def before_scenario(context: Context, scenario: Scenario): + """Setup performed before each scenario.""" + # Generate unique ID for scenario + scenario_id = uuid.uuid4() + scenario.id = scenario_id + + # Get isolated context for this scenario + scenario_context = context.scenario_context_pool.get_context(scenario_id) + scenario_context.store("test_config", BaseConfig.global_config()) + + context.logger.info(f"Preparing scenario: {scenario.name} (ID: {scenario_id})") + + def after_scenario(context: Context, scenario: Scenario): + """Cleanup performed after each scenario.""" + # Get the scenario ID + scenario_id = getattr(scenario, "id", "unknown") + context.logger.info(f"Cleaning up scenario: {scenario.name} (ID: {scenario_id})") + + # Clean up the scenario context + if hasattr(context, "scenario_context_pool"): + context.scenario_context_pool.cleanup_context(scenario_id) + +Scenario Context Management +------------------------- + +Create a scenario context to isolate test state: + +.. code-block:: python + + # features/scenario_context.py + class ScenarioContext: + """A storage class for scenario-specific objects.""" + + def __init__(self, scenario_id): + """Initialize with a unique scenario ID.""" + self.scenario_id = scenario_id + self.storage = {} + self.db_file = None + self.adapter = None + self.async_adapter = None + self.entities = {} + + def store(self, key, value): + """Store a value in the context.""" + self.storage[key] = value + + def get(self, key, default=None): + """Get a value from the context.""" + return self.storage.get(key, default) + + def cleanup(self): + """Clean up resources used by this context.""" + if self.adapter: + try: + self.adapter.session_manager.remove_session() + except Exception as e: + print(f"Error in cleanup: {e}") + + # Handle async resources too + if self.async_adapter: + import asyncio + try: + temp_loop = asyncio.new_event_loop() + try: + asyncio.set_event_loop(temp_loop) + temp_loop.run_until_complete(self.async_adapter.session_manager.remove_session()) + finally: + temp_loop.close() + except Exception as e: + print(f"Error in async cleanup: {e}") + +Context Pool Manager +------------------ + +Manage multiple scenario contexts: + +.. code-block:: python + + # features/scenario_context_pool_manager.py + from uuid import UUID + + from archipy.helpers.metaclasses.singleton import Singleton + from features.scenario_context import ScenarioContext + + class ScenarioContextPoolManager(metaclass=Singleton): + """Manager for scenario-specific context objects.""" + + def __init__(self): + """Initialize the pool manager.""" + self.context_pool = {} + + def get_context(self, scenario_id: UUID) -> ScenarioContext: + """Get or create a scenario context for the given ID.""" + if scenario_id not in self.context_pool: + self.context_pool[scenario_id] = ScenarioContext(scenario_id) + return self.context_pool[scenario_id] + + def cleanup_context(self, scenario_id: UUID) -> None: + """Clean up a specific scenario context.""" + if scenario_id in self.context_pool: + self.context_pool[scenario_id].cleanup() + del self.context_pool[scenario_id] + + def cleanup_all(self) -> None: + """Clean up all scenario contexts.""" + for scenario_id, context in list(self.context_pool.items()): + context.cleanup() + del self.context_pool[scenario_id] + +Step Implementation +----------------- + +Implementing steps with context management: + +.. code-block:: python + + # features/steps/user_steps.py + from behave import given, when, then + from sqlalchemy import select + + from archipy.adapters.orm.sqlalchemy.sqlalchemy_adapters import SqlAlchemyAdapter, AsyncSqlAlchemyAdapter + from archipy.adapters.orm.sqlalchemy.sqlalchemy_mocks import SqlAlchemyMock, AsyncSqlAlchemyMock + from features.test_helpers import get_current_scenario_context + from features.test_entity import User + + @given("the application database is initialized") + def step_given_database_initialized(context): + # Get isolated context for this scenario + scenario_context = get_current_scenario_context(context) + + # Create mock adapter for testing + adapter = SqlAlchemyMock() + scenario_context.adapter = adapter + + # Create tables + User.__table__.create(adapter.session_manager.engine) + + # Store adapter in context + scenario_context.store("adapter", adapter) + + @when('a new user "{name}" with email "{email}" is created') + def step_when_create_user(context, name, email): + scenario_context = get_current_scenario_context(context) + adapter = scenario_context.get("adapter") + + # Create user with atomic transaction + from archipy.helpers.utils.atomic_transaction import atomic_transaction + + with atomic_transaction(adapter.session_manager): + user = User(name=name, email=email) + adapter.create(user) + scenario_context.store("user", user) + + @then("the user should exist in the database") + def step_then_user_exists(context): + scenario_context = get_current_scenario_context(context) + adapter = scenario_context.get("adapter") + user = scenario_context.get("user") + + # Query for user + stored_user = adapter.get_by_uuid(User, user.test_uuid) + assert stored_user is not None + + @given('a user with email "{email}" exists') + def step_given_user_exists(context, email): + scenario_context = get_current_scenario_context(context) + + # Create async mock adapter + adapter = AsyncSqlAlchemyMock() + scenario_context.async_adapter = adapter + + # Create tables + import asyncio + + async def setup_async_db(): + await User.__table__.create(adapter.session_manager.engine) + + # Create a test user + user = User(name="Test User", email=email) + await adapter.create(user) + return user + + # Run async setup + user = asyncio.run(setup_async_db()) + scenario_context.store("async_user", user) + scenario_context.store("async_adapter", adapter) + + @when("I retrieve the user by email asynchronously") + def step_when_retrieve_user_async(context): + import asyncio + scenario_context = get_current_scenario_context(context) + adapter = scenario_context.get("async_adapter") + user = scenario_context.get("async_user") + + async def retrieve_user(): + from archipy.helpers.utils.atomic_transaction import async_atomic_transaction + + async with async_atomic_transaction(adapter.session_manager): + query = select(User).where(User.email == user.email) + users, total = await adapter.execute_search_query(User, query) + return users[0] if users else None + + # Run async retrieval + retrieved_user = asyncio.run(retrieve_user()) + scenario_context.store("retrieved_user", retrieved_user) + + @then("the user details should be retrieved successfully") + def step_then_user_details_match(context): + scenario_context = get_current_scenario_context(context) + original_user = scenario_context.get("async_user") + retrieved_user = scenario_context.get("retrieved_user") + + assert retrieved_user is not None + assert retrieved_user.test_uuid == original_user.test_uuid + assert retrieved_user.email == original_user.email + +Helper Utilities +-------------- + +ArchiPy provides test helpers for BDD scenarios: + +.. code-block:: python + + # features/test_helpers.py + from behave.runner import Context + + def get_current_scenario_context(context: Context): + """Get the current scenario context from the context pool.""" + scenario_id = context.scenario._id + return context.scenario_context_pool.get_context(scenario_id) + + class SafeAsyncContextManager: + """A safe async context manager for use in both sync and async code.""" + + def __init__(self, context_manager_factory): + self.context_manager_factory = context_manager_factory + + async def __aenter__(self): + self.context_manager = self.context_manager_factory() + return await self.context_manager.__aenter__() + + async def __aexit__(self, exc_type, exc_val, exc_tb): + return await self.context_manager.__aexit__(exc_type, exc_val, exc_tb) + +Running BDD Tests +--------------- + +Execute tests with the behave command: + +.. code-block:: bash + + # Run all tests + behave + + # Run specific feature + behave features/user_management.feature + + # Run async scenarios + behave --tags=@async + + # Generate HTML report + behave -f html -o reports/behave-report.html diff --git a/docs/source/examples/config_management.rst b/docs/source/examples/config_management.rst new file mode 100644 index 00000000..8e26482c --- /dev/null +++ b/docs/source/examples/config_management.rst @@ -0,0 +1,62 @@ +.. _examples_config: + +Configuration Management Examples +================================ + +Basic Configuration +------------------ + +.. code-block:: python + + from archipy.configs.base_config import BaseConfig + + # Define a custom configuration + class MyAppConfig(BaseConfig): + # Override defaults + AUTH = { + "SECRET_KEY": "my-secure-key", + "TOKEN_LIFETIME_MINUTES": 30 + } + + # Add custom fields + APP_NAME = "My Application" + DEBUG = True + + # Set as global configuration + config = MyAppConfig() + BaseConfig.set_global(config) + + # Access from anywhere + from archipy.configs.base_config import BaseConfig + current_config = BaseConfig.global_config() + print(current_config.APP_NAME) # "My Application" + +Environment Variables +------------------- + +ArchiPy configs automatically load from environment variables: + +.. code-block:: bash + + # .env file + AUTH__SECRET_KEY=production-secret-key + AUTH__TOKEN_LIFETIME_MINUTES=15 + APP_NAME=Production App + +Hierarchical Configuration +------------------------- + +.. code-block:: python + + from archipy.configs.base_config import BaseConfig + from archipy.configs.config_template import RedisConfig, SqlAlchemyConfig + + class MyAppConfig(BaseConfig): + # Override specific Redis settings + REDIS = RedisConfig( + MASTER_HOST="redis.example.com", + PORT=6380, + PASSWORD="redis-password" + ) + + # Keep other defaults diff --git a/docs/source/examples/helpers/decorators.rst b/docs/source/examples/helpers/decorators.rst new file mode 100644 index 00000000..ed2bf6ee --- /dev/null +++ b/docs/source/examples/helpers/decorators.rst @@ -0,0 +1,68 @@ +.. _examples_helpers_decorators: + +Decorators +========= + +Examples of ArchiPy's decorator utilities: + +retry +----- + +Automatically retry functions that might fail temporarily: + +.. code-block:: python + + from archipy.helpers.decorators.retry import retry + import requests + + # Retry up to 3 times with exponential backoff + @retry(max_attempts=3, backoff_factor=2) + def fetch_data(url): + response = requests.get(url, timeout=5) + response.raise_for_status() + return response.json() + + try: + data = fetch_data("https://api.example.com/data") + except Exception as e: + print(f"Failed after retries: {e}") + +rate_limit +--------- + +Control the frequency of function calls: + +.. code-block:: python + + from archipy.helpers.decorators.rate_limit import rate_limit + + # Limit to 5 calls per minute + @rate_limit(max_calls=5, period=60) + def send_notification(user_id, message): + # Send notification logic + print(f"Sent to {user_id}: {message}") + + # Will automatically throttle if called too frequently + for i in range(10): + send_notification(f"user_{i}", "Important update!") + +deprecation_exception +------------------- + +Mark functions as deprecated: + +.. code-block:: python + + from archipy.helpers.decorators.deprecation_exception import method_deprecation_error, class_deprecation_error + + # Method deprecation + @method_deprecation_error(operation="get_user_by_username", lang="en") + def get_user_by_username(username): + # This will raise a DeprecationError when called + pass + + # Class deprecation + @class_deprecation_error(lang="en") + class OldUserService: + # Using this class will raise a DeprecationError + pass diff --git a/docs/source/examples/helpers/index.rst b/docs/source/examples/helpers/index.rst new file mode 100644 index 00000000..7f21d28c --- /dev/null +++ b/docs/source/examples/helpers/index.rst @@ -0,0 +1,50 @@ +.. _examples_helpers: + +Helper Utilities +=============== + +ArchiPy provides various helper utilities organized by module: + +.. toctree:: + :maxdepth: 1 + :caption: Helper Modules: + + decorators + utils + metaclasses + interceptors + +Overview +-------- + +Helpers are organized into four main categories: + +* **Decorators**: Function and class decorators for cross-cutting concerns +* **Utilities**: Common utility functions for routine tasks +* **Metaclasses**: Reusable class patterns through metaclasses +* **Interceptors**: Middleware components for request processing + +Quick Examples +------------- + +.. code-block:: python + + # Retry pattern for resilient operations + from archipy.helpers.decorators.retry import retry + + @retry(max_attempts=3) + def fetch_external_data(): + # Will retry up to 3 times if this fails + pass + + # Singleton pattern for global resources + from archipy.helpers.metaclasses.singleton import Singleton + + class DatabaseConnection(metaclass=Singleton): + # Only one instance will ever exist + pass + + # Date/time utilities for consistent handling + from archipy.helpers.utils.datetime_utils import get_utc_now + + timestamp = get_utc_now() # Always UTC-aware diff --git a/docs/source/examples/helpers/interceptors.rst b/docs/source/examples/helpers/interceptors.rst new file mode 100644 index 00000000..f2f4cad9 --- /dev/null +++ b/docs/source/examples/helpers/interceptors.rst @@ -0,0 +1,48 @@ +.. _examples_helpers_interceptors: + +Interceptors +=========== + +Examples of ArchiPy's interceptor utilities: + +fastapi_rest_rate_limit_handler +----------------------------- + +Control API request rates for FastAPI: + +.. code-block:: python + + from archipy.helpers.interceptors.fastapi.rate_limit.fastapi_rest_rate_limit_handler import FastAPIRestRateLimitHandler + from fastapi import FastAPI, Request, Response + + app = FastAPI() + + # Set up rate limiting + rate_limiter = FastAPIRestRateLimitHandler( + limit=100, # 100 requests + window_seconds=60, # per minute + key_func=lambda request: request.client.host # by IP address + ) + + @app.middleware("http") + async def rate_limit_middleware(request: Request, call_next): + # Check rate limit + limited, headers = await rate_limiter.check_rate_limit(request) + + if limited: + # Rate limit exceeded + response = Response( + content="Rate limit exceeded. Try again later.", + status_code=429, + headers=headers + ) + return response + + # Continue with request + response = await call_next(request) + + # Add rate limit headers to response + for key, value in headers.items(): + response.headers[key] = value + + return response diff --git a/docs/source/examples/helpers/metaclasses.rst b/docs/source/examples/helpers/metaclasses.rst new file mode 100644 index 00000000..74176661 --- /dev/null +++ b/docs/source/examples/helpers/metaclasses.rst @@ -0,0 +1,34 @@ +.. _examples_helpers_metaclasses: + +Metaclasses +========== + +Examples of ArchiPy's metaclass utilities: + +singleton +-------- + +Ensure only one instance of a class exists: + +.. code-block:: python + + from archipy.helpers.metaclasses.singleton import Singleton + + class ConfigManager(metaclass=Singleton): + def __init__(self): + self.settings = {} + + def load_config(self, path): + # Load configuration logic + self.settings = {"key": "value"} + + # Usage + config1 = ConfigManager() + config1.load_config("config.json") + + config2 = ConfigManager() + # Both variables reference the same instance + print(config1 is config2) # True + + # Settings are shared + print(config2.settings) # {"key": "value"} diff --git a/docs/source/examples/helpers/utils.rst b/docs/source/examples/helpers/utils.rst new file mode 100644 index 00000000..79fe1c30 --- /dev/null +++ b/docs/source/examples/helpers/utils.rst @@ -0,0 +1,229 @@ +.. _examples_helpers_utils: + +Utilities +======== + +Examples of ArchiPy's utility functions: + +datetime_utils +------------- + +Work with dates and times consistently: + +.. code-block:: python + + from archipy.helpers.utils.datetime_utils import DateTimeUtils + + # Get current UTC time + now = DateTimeUtils.get_utc_now() + + # Format for storage/transmission + date_str = DateTimeUtils.convert_datetime_to_string(now) + + # Parse date string + parsed = DateTimeUtils.convert_string_to_datetime(date_str) + + # Convert to Jalali (Persian) calendar + jalali_date = DateTimeUtils.get_jalali_date(now) + + # Check if date is a holiday in Iran + is_holiday = DateTimeUtils.is_holiday_in_iran(now) + +jwt_utils +-------- + +Generate and verify JWT tokens: + +.. code-block:: python + + from archipy.helpers.utils.jwt_utils import JWTUtils + from uuid import uuid4 + + # Generate a user access token + user_id = uuid4() + access_token = JWTUtils.generate_access_token(user_id) + + # Generate a refresh token + refresh_token = JWTUtils.generate_refresh_token(user_id) + + # Verify a token + try: + payload = JWTUtils.verify_jwt(access_token) + print(f"Token valid for user: {payload['sub']}") + except Exception as e: + print(f"Invalid token: {e}") + +password_utils +------------ + +Secure password handling: + +.. code-block:: python + + from archipy.helpers.utils.password_utils import PasswordUtils + + # Hash a password + password = "SecureP@ssword123" + hashed = PasswordUtils.hash_password(password) + + # Verify password + is_valid = PasswordUtils.verify_password(password, hashed) + print(f"Password valid: {is_valid}") + + # Generate a secure password + secure_password = PasswordUtils.generate_password(length=12) + print(f"Generated password: {secure_password}") + +file_utils +--------- + +Handle files securely: + +.. code-block:: python + + from archipy.helpers.utils.file_utils import FileUtils + + # Generate secure link to file + link = FileUtils.generate_secure_file_link("/path/to/document.pdf", expires_in=3600) + + # Validate file extension + is_valid = FileUtils.validate_file_extension("document.pdf", ["pdf", "docx", "txt"]) + print(f"File is valid: {is_valid}") + +base_utils +--------- + +Validate and sanitize data: + +.. code-block:: python + + from archipy.helpers.utils.base_utils import BaseUtils + + # Sanitize phone number + phone = BaseUtils.sanitize_phone_number("+989123456789") + print(phone) # 09123456789 + + # Validate Iranian national code + try: + BaseUtils.validate_iranian_national_code_pattern("1234567891") + print("National code is valid") + except Exception as e: + print(f"Invalid national code: {e}") + +error_utils +--------- + +Standardized exception handling: + +.. code-block:: python + + from archipy.helpers.utils.error_utils import ErrorUtils + from archipy.models.errors import BaseError + from archipy.models.types.exception_message_types import ExceptionMessageType + + # Create exception detail + detail = ErrorUtils.create_exception_detail( + ExceptionMessageType.INVALID_PHONE, + lang="en" + ) + + # Handle exceptions + try: + # Some code that might fail + raise ValueError("Something went wrong") + except Exception as e: + ErrorUtils.capture_exception(e) + +app_utils +-------- + +FastAPI application utilities: + +.. code-block:: python + + from archipy.helpers.utils.app_utils import AppUtils, FastAPIUtils + from archipy.configs.base_config import BaseConfig + + # Create a FastAPI app with standard config + app = AppUtils.create_fastapi_app(BaseConfig.global_config()) + + # Add custom exception handlers + FastAPIUtils.add_exception_handlers(app) + + # Generate unique route IDs + route_id = FastAPIUtils.generate_unique_route_id("users", "get_user") + + # Set up CORS + FastAPIUtils.setup_cors( + app, + allowed_origins=["https://example.com"] + ) + +transaction_utils +---------------- + +Database transaction management: + +.. code-block:: python + + from archipy.helpers.utils.transaction_utils import TransactionUtils + from archipy.adapters.orm.sqlalchemy.session_manager_adapters import SessionManagerAdapter + + # Synchronous transaction + session_manager = SessionManagerAdapter() + + with TransactionUtils.atomic_transaction(session_manager): + # Database operations here + # Will be committed if successful, rolled back if exception occurs + pass + + # Asynchronous transaction + async with TransactionUtils.async_atomic_transaction(async_session_manager): + # Async database operations here + pass + +string_utils +---------- + +String manipulation utilities: + +.. code-block:: python + + from archipy.helpers.utils.string_utils import StringUtils + + # Convert camel case to snake case + snake = StringUtils.camel_to_snake("thisIsACamelCaseString") + print(snake) # this_is_a_camel_case_string + + # Convert snake case to camel case + camel = StringUtils.snake_to_camel("this_is_a_snake_case_string") + print(camel) # thisIsASnakeCaseString + + # Generate a random string + random_str = StringUtils.generate_random_string(length=10) + print(random_str) + + # Mask sensitive data + masked = StringUtils.mask_sensitive_data("1234567890123456", show_last=4) + print(masked) # ************3456 + +validator_utils +------------- + +Validate input data: + +.. code-block:: python + + from archipy.helpers.utils.validator_utils import ValidatorUtils + + # Validate email + is_valid_email = ValidatorUtils.is_valid_email("user@example.com") + print(f"Valid email: {is_valid_email}") + + # Validate phone number + is_valid_phone = ValidatorUtils.is_valid_phone_number("+15551234567") + print(f"Valid phone: {is_valid_phone}") + + # Validate URL + is_valid_url = ValidatorUtils.is_valid_url("https://example.com") + print(f"Valid URL: {is_valid_url}") diff --git a/docs/source/features.rst b/docs/source/features.rst index 0accc486..bbca3b8b 100644 --- a/docs/source/features.rst +++ b/docs/source/features.rst @@ -1,117 +1,53 @@ .. _features: Features -======= +======== -Modern Python Stack ------------------- +ArchiPy provides a robust framework for structured Python development, focusing on standardization, testability, and productivity. -ArchiPy is built with modern Python technologies: +Configuration Management +------------------------ -- **Python 3.13+**: Leveraging the latest Python features -- **Type Annotations**: Comprehensive type hinting for better IDE support -- **Pydantic**: For data validation and settings management -- **SQLAlchemy**: ORM for database operations -- **FastAPI**: High-performance web framework -- **gRPC**: For efficient microservice communication +- **Standardized Configs**: Use `base_config` and `config_template` for consistent setup. +- **Injection**: Seamlessly inject configurations into components. -Modular Design -------------- - -ArchiPy is designed with modularity in mind: - -- **Optional Dependencies**: Install only what you need -- **Pluggable Components**: Easy to extend and customize -- **Clean Interfaces**: Well-defined APIs between components - -Database Support --------------- - -- **SQLAlchemy Integration**: Robust ORM support -- **Connection Pooling**: Efficient database connection management -- **Async Support**: Native async database operations -- **Migration Tools**: Database schema management -- **PostgreSQL**: First-class support for PostgreSQL - -Caching with Redis +Adapters & Mocks ---------------- -- **Redis Adapter**: Easy-to-use Redis client -- **Async Support**: Async Redis operations -- **Connection Pooling**: Efficient connection management -- **Serialization**: Automatic data serialization/deserialization - -API Development -------------- - -- **FastAPI Integration**: High-performance API development -- **Request Validation**: Automatic request validation with Pydantic -- **Rate Limiting**: Built-in rate limiting support -- **Authentication**: JWT authentication support -- **Swagger Documentation**: Automatic API documentation - -gRPC Support ----------- +- **Common Adapters**: Pre-built for Redis, SQLAlchemy, and email. +- **Mocks**: Testable mocks (e.g., `redis_mocks`, `sqlalchemy_mocks`) for isolated testing. +- **Async Support**: Synchronous and asynchronous implementations. -- **Interceptors**: Client and server interceptors -- **Tracing**: Distributed tracing support -- **Metrics**: Prometheus metrics integration -- **Error Handling**: Structured error handling +Data Standardization +-------------------- -Monitoring and Observability --------------------------- +- **Base Entities**: Standardized SQLAlchemy entities (`base_entities.py`). +- **DTOs**: Pydantic-based DTOs (e.g., `pagination_dto`, `error_dto`). +- **Type Safety**: Enforced via `pydantic` and `mypy`. -- **Prometheus Integration**: Metrics collection and monitoring -- **Elastic APM**: Application performance monitoring -- **Sentry**: Error tracking and reporting -- **Structured Logging**: Consistent log formatting - -Email Support ------------ - -- **SMTP Integration**: Send emails via SMTP -- **Template Support**: Email templating -- **Attachment Handling**: Support for email attachments -- **Connection Pooling**: Efficient SMTP connection management - -Type Safety ---------- - -- **Mypy Integration**: Static type checking -- **Runtime Type Checking**: With Pydantic validation -- **Type Annotations**: Comprehensive type hints -- **Generic Types**: Support for generic types - -Testing Support -------------- +Helper Utilities +---------------- -- **Unit Testing**: With pytest -- **Integration Testing**: For testing component interactions -- **BDD Testing**: With behave for behavior-driven development -- **Mocking**: Mocking frameworks for external dependencies -- **Fixtures**: Reusable test fixtures +- **Utilities**: Tools like `datetime_utils`, `jwt_utils`, `password_utils`. +- **Decorators**: `retry`, `singleton`, `sqlalchemy_atomic`, etc. +- **Interceptors**: Rate limiting (FastAPI), tracing (gRPC). -Documentation +BDD Testing ----------- -- **Docstrings**: Comprehensive docstrings -- **Type Hints**: For better IDE support -- **Sphinx Integration**: Automatic documentation generation -- **API Documentation**: Detailed API documentation -- **Usage Examples**: Code examples for common use cases +- **Behave Integration**: Pre-configured for sync/async scenarios. +- **Feature Files**: Examples like `app_utils.feature`, `totp_utils.feature`. +- **Step Definitions**: Comprehensive steps for testing (e.g., `jwt_utils_steps.py`). -Code Quality ----------- +Best Practices & Tooling +------------------------ -- **Linting**: With ruff for code quality checks -- **Formatting**: With black for consistent code formatting -- **Type Checking**: With mypy for static type checking -- **Pre-commit Hooks**: Automated code quality checks -- **CI/CD Integration**: Continuous integration and deployment +- **Poetry**: Dependency management for reproducible builds. +- **Pre-commit**: Automated checks with `ruff`, `black`, and `mypy`. +- **Structure**: Clean architecture with `pyproject.toml` for modern Python development. -Dependency Management -------------------- +Modular Design +-------------- -- **Poetry**: Modern dependency management -- **Versioning**: Semantic versioning -- **Virtual Environments**: Isolated development environments +- **Optional Dependencies**: Install only what you need (e.g., `archipy[redis]`). +- **Extensible**: Add custom adapters and helpers easily. diff --git a/docs/source/index.rst b/docs/source/index.rst index a1d83c60..5676cd33 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -1,20 +1,56 @@ .. ArchiPy documentation master file -Welcome to ArchiPy's Documentation -================================== - -**Architecture + Python β Perfect for Structured Design** - -ArchiPy is a Python project designed to provide a robust and structured architecture for building scalable and -maintainable applications. It integrates modern Python tools and libraries to streamline development, testing, and -deployment. +Welcome to ArchiPy +================= .. image:: https://img.shields.io/badge/python-3.13+-blue.svg :target: https://www.python.org/downloads/ :alt: Python 3.13+ +.. image:: ../assets/logo.jpg + :alt: ArchiPy Logo + :width: 150 + :align: right + +**Architecture + Python β Structured Development Simplified** + +ArchiPy provides a clean architecture framework for Python applications that: + +* Standardizes configuration management +* Offers pluggable adapters with testing mocks +* Enforces consistent data models +* Promotes maintainable code organization +* Simplifies testing with BDD support + +.. grid:: 2 + + .. grid-item-card:: π Quick Start + :link: installation + :link-type: ref + + Get started with ArchiPy in minutes + + .. grid-item-card:: π Features + :link: features + :link-type: ref + + Explore what ArchiPy offers + + .. grid-item-card:: ποΈ Architecture + :link: architecture + :link-type: ref + + Learn about the design principles + + .. grid-item-card:: π API Reference + :link: api_reference/index + :link-type: ref + + Detailed API documentation + .. toctree:: :maxdepth: 2 + :hidden: :caption: Contents: installation @@ -27,6 +63,16 @@ deployment. changelog license +.. toctree:: + :maxdepth: 1 + :hidden: + :caption: Examples: + + examples/config_management + examples/adapters/index + examples/helpers/index + examples/bdd_testing + Quick Start ---------- @@ -38,7 +84,7 @@ Quick Start # Or with poetry poetry add archipy -Indices and tables +Indices and Tables ================== * :ref:`genindex` diff --git a/docs/source/installation.rst b/docs/source/installation.rst index e4545197..262e440a 100644 --- a/docs/source/installation.rst +++ b/docs/source/installation.rst @@ -4,62 +4,61 @@ Installation =========== Prerequisites ------------- +------------- -Before you begin, ensure you have the following installed: +Before starting, ensure you have: - **Python 3.13 or higher** - ArchiPy is compatible with Python 3.13 and above but does not support Python 4 or higher. - To check your Python version, run: + ArchiPy requires Python 3.13+. Check your version with: .. code-block:: bash python --version - If your Python version is lower than 3.13, `download and install the latest version of Python