### Quick start example for pydynox Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/index.md Basic implementation demonstrating how to integrate pydynox within an agentic workflow. ```python --8<-- "docs/examples/agentic/quick_example.py" ``` -------------------------------- ### Interop Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/migrating-from-boto3.md Provides an example of how to use both boto3 and pydynox within the same Python script during a migration process. ```python --8<-- "docs/examples/migration/interop.py" ``` -------------------------------- ### Clone Repository and Setup Virtual Environment Source: https://github.com/ferrumio/pydynox/blob/main/CONTRIBUTING.md Clones the pydynox repository and sets up a Python virtual environment. Installs maturin and development dependencies, including the project in editable mode. ```bash git clone https://github.com/yourusername/pydyno.git cd pydyno # Create virtual environment python -m venv .venv source .venv/bin/activate # On Windows: .venv\Scripts\activate # Install maturin and dev dependencies pip install maturin pip install -e ".[dev]" ``` -------------------------------- ### Install pydynox and dependencies Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/smolagents.md Install the necessary packages to use pydynox with Smolagents and LiteLLM. ```bash pip install pydynox smolagents litellm ``` -------------------------------- ### Install pydynox Source: https://github.com/ferrumio/pydynox/blob/main/README.md Install the core library and optional extras for Pydantic or OpenTelemetry support. ```bash pip install pydynox ``` ```bash pip install pydynox[pydantic] # Pydantic integration pip install pydynox[opentelemetry] # OpenTelemetry tracing ``` -------------------------------- ### Install Rust using rustup Source: https://github.com/ferrumio/pydynox/blob/main/CONTRIBUTING.md Installs the latest stable version of Rust using the official rustup installer. This is a prerequisite for building and testing Rust components. ```bash curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh ``` -------------------------------- ### Install pydynox and pydantic-ai Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/pydantic-ai.md Install the necessary libraries for using Pydantic AI with pydynox. ```bash pip install pydynox pydantic-ai ``` -------------------------------- ### Install pydynox Source: https://github.com/ferrumio/pydynox/blob/main/docs/getting-started.md Installation commands for common Python package managers. ```bash pip install pydynox ``` ```bash uv add pydynox ``` -------------------------------- ### Install OpenTelemetry Dependency Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/observability.md Install the optional OpenTelemetry support for pydynox. ```bash pip install pydynox[opentelemetry] ``` -------------------------------- ### Hot Partition Detection Log Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/diagnostics/hot-partition.md This is an example of a log message generated when a hot partition is detected. ```log WARNING:pydynox.diagnostics:Hot partition detected - table="events" pk="EVENTS" had 500 writes in 60s ``` -------------------------------- ### Verify pydynox installation Source: https://github.com/ferrumio/pydynox/blob/main/docs/getting-started.md Check the installed version and environment details. ```python import pydynox print(pydynox.__version__) # 0.1.0 # For detailed version info print(pydynox.version_info()) # pydynox version: 0.1.0 # python version: 3.11.0 # platform: macOS-14.0-arm64 # related packages: boto3-1.34.0 pydantic-2.5.0 ``` -------------------------------- ### Install Pydynox and Strands Agents Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/strands.md Install the necessary packages to enable pydynox integration with Strands Agents. ```bash pip install pydynox strands-agents ``` -------------------------------- ### Basic Hot Partition Detection Setup Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/diagnostics/hot-partition.md Initialize the HotPartitionDetector and pass it to your client. Warnings will be logged when partition thresholds are exceeded. ```python from pydynox.diagnostics import HotPartitionDetector from pydynox.client import Client # Initialize the detector with thresholds and window duration detector = HotPartitionDetector( writes_threshold=500, reads_threshold=1500, window_seconds=60 ) # Pass the detector to your client client = Client(..., diagnostics=detector) ``` -------------------------------- ### Install pydynox with Pydantic support Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/pydantic.md Install the pydynox library with the necessary extras for Pydantic integration using pip. ```bash pip install pydynox[pydantic] ``` -------------------------------- ### Full Pydantic AI Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/pydantic-ai.md A comprehensive example demonstrating the integration of Pydantic AI with pydynox. ```python --8<-- "docs/examples/agentic/pydantic_ai_tools.py" ``` -------------------------------- ### Implement document management model Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/s3-attribute.md Complete example showing model definition, file upload, querying, and generating a presigned URL. ```python from pathlib import Path from pydynox import Model, ModelConfig from pydynox.attributes import StringAttribute, S3Attribute, DatetimeAttribute, AutoGenerate from pydynox._internal._s3 import S3File class Document(Model): model_config = ModelConfig(table="documents") pk = StringAttribute(partition_key=True) sk = StringAttribute(sort_key=True, default="v1") name = StringAttribute() uploaded_at = DatetimeAttribute(default=AutoGenerate.utc_now()) content = S3Attribute(bucket="company-docs", prefix="documents/") # Upload a new document doc = Document(pk="DOC#invoice-2024-001", name="Invoice January 2024") doc.content = S3File( Path("/tmp/invoice.pdf"), content_type="application/pdf", metadata={"department": "finance"} ) await doc.save() # List documents (fast, no S3 calls) async for doc in Document.query(partition_key="DOC#invoice-2024-001"): print(f"{doc.name}: {doc.content.size} bytes") # Generate download link url = await doc.content.presigned_url(expires=3600) print(f"Download: {url}") ``` -------------------------------- ### Deploy Pydynox Benchmarks with CDK Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/benchmarks.md Use these commands to set up the benchmarking environment. Ensure you are in the 'benchmarks' directory and have installed the necessary Python dependencies. ```bash cd benchmarks pip install -r requirements.txt cdk deploy ``` -------------------------------- ### Run DynamoDB Local Source: https://github.com/ferrumio/pydynox/blob/main/docs/getting-started.md Start a local DynamoDB instance using Docker. ```bash docker run -p 8000:8000 amazon/dynamodb-local ``` -------------------------------- ### Define a basic model Source: https://github.com/ferrumio/pydynox/blob/main/docs/getting-started.md Example of a basic User model definition. ```python --8<-- "docs/examples/models/basic_model.py" ``` -------------------------------- ### Configure OpenTelemetry Tracing Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/observability.md Examples for basic usage, custom configuration, and disabling tracing. ```python --8<-- "docs/examples/observability/tracing_basic.py" ``` ```python --8<-- "docs/examples/observability/tracing_custom.py" ``` ```python --8<-- "docs/examples/observability/tracing_disable.py" ``` -------------------------------- ### Complete Single-Table Design Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/inheritance.md A comprehensive example demonstrating single-table design using Pydynox inheritance, including a base class for audit fields and a parent model with a discriminator. ```python from pydynox import Model, ModelConfig from pydynox.attributes import StringAttribute, NumberAttribute, BooleanAttribute # Base class for audit fields class AuditBase: created_by = StringAttribute() # Parent model - all entities in one table class Entity(Model, AuditBase): model_config = ModelConfig(table="app") pk = StringAttribute(partition_key=True) sk = StringAttribute(sort_key=True) entity_type = StringAttribute(discriminator=True) # User entity class User(Entity): model_config = ModelConfig(table="app") name = StringAttribute() email = StringAttribute() # Product entity class Product(Entity): model_config = ModelConfig(table="app") title = StringAttribute() price = NumberAttribute() # Order entity class Order(Entity): model_config = ModelConfig(table="app") user_id = StringAttribute() total = NumberAttribute() ``` -------------------------------- ### Basic Types Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/type-checking.md An example demonstrating basic attribute types in pydynox. ```python from datetime import datetime from typing import Any, Dict, List, Set from pydynox.attributes import ( BinaryAttribute, BooleanAttribute, CompressedAttribute, DatetimeAttribute, EncryptedAttribute, JSONAttribute, ListAttribute, MapAttribute, ModelAttribute, NumberAttribute, NumberSetAttribute, S3Attribute, StringSetAttribute, StringAttribute, TTLAttribute, VersionAttribute, ) from pydynox.model import Model class MyModel(Model): pk: str sk: str class ExampleModel(Model): string_attr: StringAttribute = StringAttribute() number_attr: NumberAttribute = NumberAttribute() boolean_attr: BooleanAttribute = BooleanAttribute() binary_attr: BinaryAttribute = BinaryAttribute() list_attr: ListAttribute = ListAttribute() map_attr: MapAttribute = MapAttribute() string_set_attr: StringSetAttribute = StringSetAttribute() number_set_attr: NumberSetAttribute = NumberSetAttribute() json_attr: JSONAttribute = JSONAttribute() json_model_attr: JSONAttribute[MyModel] = JSONAttribute(MyModel) datetime_attr: DatetimeAttribute = DatetimeAttribute() ttl_attr: TTLAttribute = TTLAttribute() compressed_attr: CompressedAttribute = CompressedAttribute() encrypted_attr: EncryptedAttribute = EncryptedAttribute() s3_attr: S3Attribute = S3Attribute() version_attr: VersionAttribute = VersionAttribute() ``` -------------------------------- ### Create Table and Wait Separately (Sync) Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/tables.md Create a table synchronously and then explicitly wait for it to become active using `sync_wait_for_table_active`. This allows for other setup operations in between. ```python client.sync_create_table("users", partition_key=("pk", "S")) # Do other setup... client.sync_wait_for_table_active("users", timeout_seconds=30) ``` -------------------------------- ### Implement Session Management with TTL Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/ttl.md A practical example of using TTL for managing session lifecycles. ```python --8<-- "docs/examples/ttl/session_example.py" ``` -------------------------------- ### Async and Sync key operations Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/models.md Examples of performing updates and deletes by key without fetching the item first. ```python --8<-- "docs/examples/models/key_operations.py" ``` ```python --8<-- "docs/examples/models/sync_key_operations.py" ``` -------------------------------- ### Create Table and Wait Separately (Async) Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/tables.md Create a table asynchronously and then explicitly wait for it to become active using `wait_for_table_active`. This allows for other setup operations in between. ```python await client.create_table("users", partition_key=("pk", "S")) # Do other setup... await client.wait_for_table_active("users", timeout_seconds=30) ``` -------------------------------- ### Define Smolagents tools Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/smolagents.md Example implementation of tools for Smolagents using pydynox. ```python --8<-- "docs/examples/agentic/smolagents_tools.py" ``` -------------------------------- ### Create Table and Wait (Sync) Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/tables.md Use `wait=True` to block until the table is ready for use. This example demonstrates creating a table using both the Model and client interfaces synchronously. ```python # Using Model User.sync_create_table(wait=True) # Table is now ready to use # Using client client.sync_create_table("users", partition_key=("pk", "S"), wait=True) ``` -------------------------------- ### Implement CRUD Tools for Agents Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/strands.md Example showing how to structure CRUD operations as tools for Strands Agents. ```python --8<-- "docs/examples/agentic/strands_crud.py" ``` -------------------------------- ### Define Strands Tools Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/strands.md Example implementation of tools for Strands Agents using pydynox. ```python --8<-- "docs/examples/agentic/strands_tools.py" ``` -------------------------------- ### Build pydynox Locally Source: https://github.com/ferrumio/pydynox/blob/main/CONTRIBUTING.md Builds and installs the pydynox project in development mode using maturin. Also shows how to build the release version. ```bash # Build and install in development mode maturin develop # Build release version maturin build --release ``` -------------------------------- ### Create Table and Wait (Async) Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/tables.md Use `wait=True` to block until the table is ready for use. This example demonstrates creating a table using both the Model and client interfaces asynchronously. ```python await User.create_table(wait=True) # Table is now ready to use # Using client await client.create_table("users", partition_key=("pk", "S"), wait=True) ``` -------------------------------- ### Benchmark update pattern Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/benchmarks.md Example of the current update pattern that requires a prior fetch, which is currently being optimized. ```python item = Model.get(pk="...") # GET first item.update(data="new") # then UPDATE ``` -------------------------------- ### Use Cases for Logging Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/observability.md Examples demonstrating how to leverage pydynox's logging capabilities for cost monitoring, performance debugging, and Lambda optimization. ```APIDOC ## Use Cases ### Cost Monitoring Track capacity consumption per operation: ```python # For Model operations user = User.get(pk="USER#123", sk="PROFILE") last = User.get_last_metrics() if last: print(f"This read cost {last.consumed_rcu} RCU") # For client operations client.get_item("users", {"pk": "USER#123"}) last = client.get_last_metrics() if last: print(f"This read cost {last.consumed_rcu} RCU") ``` ### Performance Debugging Find slow operations: ```python for order in Order.query(partition_key="CUSTOMER#123"): print(order.total) last = Order.get_last_metrics() if last and last.duration_ms > 100: logger.warning(f"Slow query: {last.duration_ms}ms") ``` ### Lambda Optimization In Lambda, every millisecond counts: ```python def handler(event, context): set_correlation_id(context.aws_request_id) user = User.get(pk=event["user_id"]) # Logs include request ID for tracing return {"statusCode": 200} ``` ``` -------------------------------- ### Synchronous Collection Query Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/single-table.md Provides an example of performing a synchronous query on a collection using `sync_query`. ```python result = collection.sync_query(pk="USER#123") ``` -------------------------------- ### Run Integration Tests Source: https://github.com/ferrumio/pydynox/blob/main/ADR/006-unit-vs-integration-tests.md Command to execute only the integration tests. This suite requires a localstack environment or similar setup. ```bash uv run pytest tests/integration/ ``` -------------------------------- ### Configure GSI Projection Types Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/indexes.md Examples for defining index projections using KEYS_ONLY or INCLUDE types. ```python # Keys only - smallest index, lowest cost { "index_name": "status-index", "partition_key": ("status", "S"), "projection": "KEYS_ONLY", } # Include specific attributes { "index_name": "email-index", "partition_key": ("email", "S"), "projection": "INCLUDE", "non_key_attributes": ["name", "created_at"], } ``` -------------------------------- ### Async CRUD with S3 using Pydynox Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/pydantic-ai.md Example showcasing asynchronous Create, Read, Update, and Delete operations with S3 storage using pydynox. ```python --8<-- "docs/examples/agentic/pydantic_ai_s3.py" ``` -------------------------------- ### Track KMS Metrics Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/operations-metrics.md Example demonstrating how to track KMS metrics when using EncryptedAttribute. This helps in understanding the latency and cost added by KMS calls during save and get operations. ```python --8< "docs/examples/observability/kms_metrics.py" ``` -------------------------------- ### Define Multi-attribute Keys Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/indexes.md Examples of using lists for partition and sort keys to handle multi-tenant, hierarchical, or time-series data without synthetic keys. ```python partition_key=["tenant_id", "entity_type"] ``` ```python partition_key=["country", "state"] ``` ```python sort_key=["year", "month", "day"] ``` ```python sort_key=["priority", "created_at"] ``` -------------------------------- ### Create a table with client Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/tables.md Create tables directly using the client with explicit partition and sort keys. ```python --8<-- "docs/examples/tables/create_table_async.py" ``` ```python --8<-- "docs/examples/tables/create_table.py" ``` -------------------------------- ### Build Pydynox from Source Source: https://github.com/ferrumio/pydynox/blob/main/README.md Commands to clone, build, and test the project from source. ```bash # Clone git clone https://github.com/ferrumio/pydynox.git cd pydynox # Build (requires Python 3.11+, Rust 1.70+) pip install maturin maturin develop # Test pip install -e ".[dev]" pytest ``` -------------------------------- ### Create Client Table Async Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/async-first.md Use `client.create_table()` for asynchronous table creation via a client instance. Requires an initialized client. ```python await client.create_table() ``` -------------------------------- ### Track S3 Metrics Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/operations-metrics.md Example showing how to track S3 metrics when using S3Attribute. This is useful for monitoring latency and data transfer costs associated with S3 uploads and downloads. ```python --8< "docs/examples/observability/s3_metrics.py" ``` -------------------------------- ### client.create_table() / client.sync_create_table() Parameters Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/tables.md Detailed explanation of parameters for creating tables using Client methods. ```APIDOC ## client.create_table() / client.sync_create_table() Parameters | Parameter | Type | Default | Description | |---|---|---|---| | `table_name` | str | Required | Name of the table | | `partition_key` | tuple | Required | (name, type) for partition key | | `sort_key` | tuple | None | (name, type) for sort key | | `billing_mode` | str | `"PAY_PER_REQUEST"` | Billing mode | | `read_capacity` | int | 5 | RCU (only for PROVISIONED) | | `write_capacity` | int | 5 | WCU (only for PROVISIONED) | | `table_class` | str | `"STANDARD"` | Storage class | | `encryption` | str | `"AWS_OWNED"` | Encryption type | | `kms_key_id` | str | None | KMS key ARN | | `global_secondary_indexes` | list | None | GSI definitions | | `wait` | bool | False | Wait for table to be active | ``` -------------------------------- ### Pydantic Model Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/type-checking.md An example showing how Pydantic models work with pydynox. ```python from pydantic import BaseModel from pydynox.integrations.pydantic import dynamodb_model @dynamodb_model(table="products", partition_key="pk") class Product(BaseModel): pk: str name: str price: float product = Product(pk="PROD#1", name="Widget", price=9.99) print(product.pk) print(product.name) print(product.price) ``` -------------------------------- ### Configure default and model-specific clients Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/models.md Shows how to set a global default client or override it for specific models. ```python from pydynox import DynamoDBClient, set_default_client # At app startup client = DynamoDBClient(region="us-east-1", profile="prod") set_default_client(client) # All models use this client class User(Model): model_config = ModelConfig(table="users") pk = StringAttribute(partition_key=True) class Order(Model): model_config = ModelConfig(table="orders") pk = StringAttribute(partition_key=True) ``` ```python # Default client for most models set_default_client(prod_client) # Special client for audit logs audit_client = DynamoDBClient(region="eu-west-1") class AuditLog(Model): model_config = ModelConfig( table="audit_logs", client=audit_client, # Uses different client ) pk = StringAttribute(partition_key=True) ``` -------------------------------- ### Manual Key Construction with Potential Errors Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/single-table.md Demonstrates manual key construction for different entity types. Without template keys, it's easy to create an Order with a User's key pattern, leading to data corruption. ```python class User(Model): model_config = ModelConfig(table="app") pk = StringAttribute(partition_key=True) sk = StringAttribute(sort_key=True) class Order(Model): model_config = ModelConfig(table="app") pk = StringAttribute(partition_key=True) sk = StringAttribute(sort_key=True) # Bug: Order with User's key pattern - no error! Order(pk="USER#john@example.com", sk="PROFILE") ``` -------------------------------- ### Handling None Values Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/type-checking.md An example demonstrating how to handle None values explicitly in pydynox. ```python from typing import cast, Optional from pydynox.model import Model class User(Model): pk: str sk: str name: Optional[str] = None age: Optional[int] = None def process_user(user: User): # Using assertion assert user.name is not None print(f"User name: {user.name.upper()}") # Using narrowing if user.age is not None: print(f"User age: {user.age * 2}") # Using cast for query results where as_dict=False is guaranteed # Assuming 'get_user_from_db' returns User or None db_user = get_user_from_db(user.pk) if db_user: # If you are certain db_user is not None and is a User instance # and you want to assert this for type checkers confirmed_user = cast(User, db_user) print(f"Confirmed user name: {confirmed_user.name}") def get_user_from_db(pk: str) -> Optional[User]: # Placeholder for actual database retrieval logic # This might return a User object or None return User(pk=pk, sk='SK') # Example return ``` -------------------------------- ### S3Attribute Basic Upload Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/attributes.md Shows a basic example of uploading a file using the S3Attribute. ```python --8<-- "docs/examples/s3/basic_upload.py" ``` -------------------------------- ### CRUD Method Return Types Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/type-checking.md An example illustrating the return types of CRUD methods in pydynox. ```python from typing import Any, Dict, List, Optional, Set from pydynox.model import Model class User(Model): pk: str sk: str name: str age: int def create(self) -> None: ... def save(self) -> None: ... def get(self) -> Optional['User']: ... # Returns User or None @classmethod def get_by_pk(cls, pk: str) -> Optional['User']: ... # Returns User or None @classmethod def query(cls, **kwargs: Any) -> List['User']: ... # Returns a list of User @classmethod def scan(cls, **kwargs: Any) -> List['User']: ... # Returns a list of User ``` -------------------------------- ### Get detailed item size breakdown Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/size-calculator.md Pass `detailed=True` to `calculate_size()` to get a per-field breakdown of the item's size. This helps identify which fields are consuming the most space. ```python from pydynox.models import Model class Post(Model): pk: str sk: str title: str body: str tags: list[str] post = Post( pk="POST#1", sk="POST#1", title="Hello world", body="This is the post body.", tags=["hello", "world"], ) size = post.calculate_size(detailed=True) print(f"Total: {size.bytes} bytes") print("Per field:") for name, field_size in size.fields.items(): print(f" {name}: {field_size} bytes") ``` -------------------------------- ### Query data by department Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/smolagents.md Example of querying database records filtered by department. ```python --8<-- "docs/examples/agentic/smolagents_department.py" ``` -------------------------------- ### Basic Collection Usage Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/single-table.md Demonstrates the fundamental way to initialize and use a Pydynox collection. ```python --8<-- "docs/examples/collection/basic.py" ``` -------------------------------- ### Get Remaining Time Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/ttl.md Retrieve the remaining time until expiration as a timedelta object. ```python remaining = session.expires_in if remaining: print(f"Expires in {remaining.total_seconds()} seconds") print(f"Expires in {remaining.total_seconds() / 60} minutes") else: print("Already expired or no TTL") ``` -------------------------------- ### DynamoDB metadata structure Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/s3-attribute.md Example of the JSON structure stored in DynamoDB for an S3-backed attribute. ```json { "pk": {"S": "DOC#1"}, "name": {"S": "report.pdf"}, "content": { "M": { "bucket": {"S": "my-bucket"}, "key": {"S": "docs/DOC/1/report.pdf"}, "size": {"N": "1048576"}, "etag": {"S": "d41d8cd98f00b204e9800998ecf8427e"}, "content_type": {"S": "application/pdf"} } } } ``` -------------------------------- ### Perform unsafe read-modify-write Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/atomic-updates.md Example of a race condition prone update pattern that should be avoided. ```python # Dangerous - race condition! user = User.get(pk="USER#123") user.login_count = user.login_count + 1 user.save() ``` -------------------------------- ### Using Dependencies in Pydynox Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/agentic/pydantic-ai.md Demonstrates how to manage and use dependencies within pydynox tools, leveraging the context object. ```python --8<-- "docs/examples/agentic/pydantic_ai_deps.py" ``` -------------------------------- ### Get Total Metrics Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/query.md Retrieves and prints the total RCU consumed across all operations. ```python total = Order.get_total_metrics() print(f"Total RCU: {total.total_rcu}") ``` -------------------------------- ### reset_metrics() Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/operations-metrics.md Resets the accumulated metrics counters, typically used at the start of a request cycle. ```APIDOC ## POST reset_metrics() ### Description Clears all current metric counters for the model to ensure accurate tracking for a new operation or request context. ``` -------------------------------- ### Querying with Template Placeholders Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/single-table.md Shows how to query using template placeholder names instead of manually constructing keys. This approach is cleaner and less prone to errors. ```python # Without template - you build the key yourself async for order in Order.query(partition_key="USER#123"): print(order) # With template - pass the placeholder value async for order in Order.query(user_id="123"): print(order) ``` -------------------------------- ### Perform atomic update Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/atomic-updates.md Example of using the atomic parameter to safely increment a value in DynamoDB. ```python # Safe - no race condition user.update(atomic=[User.login_count.add(1)]) ``` -------------------------------- ### Table Creation with Wait Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/tables.md Demonstrates how to create DynamoDB tables and optionally wait for them to become active. ```APIDOC ## Table Creation with Wait DynamoDB tables take time to provision. The `wait=True` parameter can be used to block execution until the table is ready. ### Async (default) ```python # Using Model await User.create_table(wait=True) # Table is now ready to use # Using client await client.create_table("users", partition_key=("pk", "S"), wait=True) ``` ### Sync ```python # Using Model User.sync_create_table(wait=True) # Table is now ready to use # Using client client.sync_create_table("users", partition_key=("pk", "S"), wait=True) ``` ### Separate Wait Alternatively, you can initiate table creation and then wait for its activation separately. #### Async (default) ```python await client.create_table("users", partition_key=("pk", "S")) # Do other setup... await client.wait_for_table_active("users", timeout_seconds=30) ``` #### Sync ```python client.sync_create_table("users", partition_key=("pk", "S")) # Do other setup... client.sync_wait_for_table_active("users", timeout_seconds=30) ``` ``` -------------------------------- ### Create Client Table Sync Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/async-first.md Use `client.sync_create_table()` for synchronous table creation via a client. This method blocks execution. ```python client.sync_create_table() ``` -------------------------------- ### Detailed Item Size Breakdown Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/size-calculator.md Get a per-field breakdown of item size to identify large fields. ```APIDOC ## POST /calculate_size?detailed=True ### Description Calculates the size of a model instance and provides a detailed breakdown per field. ### Method POST ### Endpoint /calculate_size ### Parameters #### Query Parameters - **detailed** (bool) - Required - Set to `True` to enable detailed breakdown. ### Request Body - **model_instance** (object) - Required - The model instance to calculate the size for. ### Request Example ```json { "model_instance": { "pk": "POST#1", "sk": "COMMENT#1", "title": "My First Post", "body": "This is the content of my first post.", "metadata": {"timestamp": "2023-01-01T10:00:00Z"}, "tags": ["python", "dynamodb"] } } ``` ### Response #### Success Response (200) - **total_bytes** (int) - Total size of the item in bytes. - **fields** (dict) - A dictionary where keys are field names and values are their sizes in bytes. #### Response Example ```json { "total_bytes": 10089, "fields": { "body": 10003, "metadata": 42, "tags": 24, "title": 15, "pk": 5, "sk": 5 } } ``` ``` -------------------------------- ### Get Item Sync Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/async-first.md Use `client.sync_get_item()` for synchronous retrieval of a single item. This method blocks execution. ```python client.sync_get_item() ``` -------------------------------- ### Model.create_table() / Model.sync_create_table() Parameters Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/tables.md Detailed explanation of parameters for creating tables using Model methods. ```APIDOC ## Model.create_table() / Model.sync_create_table() Parameters | Parameter | Type | Default | Description | |---|---|---|---| | `billing_mode` | str | `"PAY_PER_REQUEST"` | Billing mode | | `read_capacity` | int | None | RCU (only for PROVISIONED) | | `write_capacity` | int | None | WCU (only for PROVISIONED) | | `table_class` | str | None | Storage class | | `encryption` | str | None | Encryption type | | `kms_key_id` | str | None | KMS key ARN | | `wait` | bool | False | Wait for table to be active | ``` -------------------------------- ### Get AWS Region Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/client.md Use the `get_region()` method to retrieve the AWS region the client is configured to use. ```python from pydynox.client import Client client = Client(region="us-east-1") region = client.get_region() print(f"Client is configured for region: {region}") ``` -------------------------------- ### Read items Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/models.md Retrieve items using the get() class method, checking for None if the item does not exist. ```python # Async user = await User.get(pk="USER#123", sk="PROFILE") if user: print(user.name) else: print("User not found") # Sync user = User.sync_get(pk="USER#123", sk="PROFILE") ``` ```python user = await User.get(pk="USER#123") ``` -------------------------------- ### Configure model connection Source: https://github.com/ferrumio/pydynox/blob/main/docs/getting-started.md Set up table and connection parameters using ModelConfig. ```python from pydynox import Model, ModelConfig class User(Model): model_config = ModelConfig( table="users", # Required - table name region="us-east-1", # Optional - AWS region endpoint_url=None, # Optional - for local testing ) ``` -------------------------------- ### Python Type Hinting Example Source: https://github.com/ferrumio/pydynox/blob/main/CONTRIBUTING.md Demonstrates correct Python type hinting for function signatures, adhering to PEP 8 and project standards. Use type hints for better code clarity and maintainability. ```python # Good def get_user(user_id: str) -> Optional[User]: ... # Bad def get_user(user_id): ... ``` -------------------------------- ### Get Item Comparison Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/migrating-from-boto3.md Compares how to retrieve a single item from a DynamoDB table using boto3 and pydynox. ```python --8<-- "docs/examples/migration/get_item_boto3.py" ``` ```python --8<-- "docs/examples/migration/get_item_pydynox.py" ``` -------------------------------- ### Run Unit Tests Source: https://github.com/ferrumio/pydynox/blob/main/ADR/006-unit-vs-integration-tests.md Command to execute only the unit tests. This is the faster test suite and does not require Docker. ```bash uv run pytest tests/unit/ ``` -------------------------------- ### Complete IAM Policy Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/iam-permissions.md A combined policy for a service performing CRUD, batch operations, and field encryption. ```json { "Version": "2012-10-17", "Statement": [ { "Sid": "DynamoDBAccess", "Effect": "Allow", "Action": [ "dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:DeleteItem", "dynamodb:Query", "dynamodb:Scan", "dynamodb:BatchGetItem", "dynamodb:BatchWriteItem" ], "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/users" }, { "Sid": "KMSAccess", "Effect": "Allow", "Action": [ "kms:GenerateDataKey", "kms:Decrypt" ], "Resource": "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012" } ] } ``` -------------------------------- ### Create Table Async Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/async-first.md Use `Model.create_table()` for asynchronous table creation. Ensure the model is defined before calling. ```python await Model.create_table() ``` -------------------------------- ### Get Last Operation Metrics Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/query.md Retrieves and prints metrics for the last operation. Requires the 'last' object to be truthy. ```python last = Order.get_last_metrics() if last: print(f"Duration: {last.duration_ms}ms") print(f"RCU consumed: {last.consumed_rcu}") ``` -------------------------------- ### Implement basic versioning with pydynox Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/optimistic-locking.md Basic implementation of the VersionAttribute for tracking item versions. ```python --8<-- "docs/examples/optimistic_locking/basic_version.py" ``` -------------------------------- ### EncryptedAttribute Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/attributes.md Demonstrates how to use EncryptedAttribute to encrypt sensitive data using AWS KMS. ```python from pydynox.attributes import EncryptedAttribute, EncryptionMode class User(Model): model_config = ModelConfig(table="users") pk = StringAttribute(partition_key=True) ssn = EncryptedAttribute(key_id="alias/my-key") ``` -------------------------------- ### CompressedAttribute Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/attributes.md Shows how to use CompressedAttribute for auto-compressing large text data with different algorithms. ```python from pydynox.attributes import CompressedAttribute, CompressionAlgorithm class Document(Model): model_config = ModelConfig(table="documents") pk = StringAttribute(partition_key=True) body = CompressedAttribute() # Uses zstd by default logs = CompressedAttribute(algorithm=CompressionAlgorithm.Lz4) # Example usage doc = Document(pk="DOC#1", body="This is a large text body...") await doc.save() doc_with_logs = Document(pk="DOC#2", logs="Log entry 1\nLog entry 2") await doc_with_logs.save() ``` -------------------------------- ### Basic Sync Query Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/query.md Use `Model.sync_query()` to fetch items by hash key synchronously. The result can be iterated over or accessed for the first/all items. ```python from pydynox.models import Model class Order(Model): pass def main(): # Fetch items by hash key for order in Order.sync_query(partition_key="CUSTOMER#123"): print(order.sk) # Get first result first_order = Order.sync_query(partition_key="CUSTOMER#123").first() print(first_order.sk) # Collect all results all_orders = list(Order.sync_query(partition_key="CUSTOMER#123")) print(len(all_orders)) ``` -------------------------------- ### DatetimeAttribute Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/attributes.md Demonstrates storing datetime objects as ISO 8601 strings using DatetimeAttribute. ```python from datetime import datetime from pydynox.attributes import DatetimeAttribute class LogEntry(Model): model_config = ModelConfig(table="log_entries") pk = StringAttribute(partition_key=True) timestamp = DatetimeAttribute() # Store a datetime object log = LogEntry(pk="LOG#1", timestamp=datetime.utcnow()) await log.save() # Naive datetimes are treated as UTC log_naive = LogEntry(pk="LOG#2", timestamp=datetime(2023, 1, 1, 12, 0, 0)) await log_naive.save() ``` -------------------------------- ### Run All Tests Source: https://github.com/ferrumio/pydynox/blob/main/ADR/006-unit-vs-integration-tests.md Command to execute all tests, including both unit and integration tests. This requires Docker to be running for the integration tests. ```bash uv run pytest ``` -------------------------------- ### Connect to DynamoDB Local Source: https://github.com/ferrumio/pydynox/blob/main/docs/getting-started.md Configure the model to point to a local DynamoDB instance. ```python from pydynox import Model, ModelConfig class User(Model): model_config = ModelConfig( table="users", endpoint_url="http://localhost:8000", ) ``` -------------------------------- ### EnumAttribute Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/attributes.md Illustrates storing Python enums using EnumAttribute for type-safe value storage. ```python from enum import Enum from pydynox.attributes import EnumAttribute class Status(Enum): PENDING = "pending" COMPLETED = "completed" FAILED = "failed" class Task(Model): model_config = ModelConfig(table="tasks") pk = StringAttribute(partition_key=True) status = EnumAttribute(enum_class=Status) task = Task(pk="TASK#1", status=Status.PENDING) await task.save() print(task.status == Status.PENDING) # True ``` -------------------------------- ### JSONAttribute Example Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/attributes.md Demonstrates how to use JSONAttribute to store dictionary or list data as a JSON string. ```python from pydynox.attributes import JSONAttribute class Event(Model): model_config = ModelConfig(table="events") pk = StringAttribute(partition_key=True) payload = JSONAttribute() # Store a dict event = Event(pk="EVENT#1", payload={"score": 0.99, "user": "alice"}) await event.save() # Store a list event = Event(pk="EVENT#2", payload=[1, 2, 3]) await event.save() ``` -------------------------------- ### basic.py Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/smart-save.md Demonstrates the basic usage of smart save in pydynox. ```python --8<-- "docs/examples/smart_save/basic.py" ``` -------------------------------- ### Get S3 Bytes Sync Source: https://github.com/ferrumio/pydynox/blob/main/docs/guides/async-first.md Use `s3_value.sync_get_bytes()` to synchronously retrieve the bytes of an S3 attribute. This method blocks execution. ```python s3_value.sync_get_bytes() ```