| Name | fullmetalalchemy JSON |
| Version |
2.2.3
JSON |
| download |
| home_page | None |
| Summary | Easy-to-use helpers for SQL table changes with SQLAlchemy. |
| upload_time | 2025-10-18 19:13:15 |
| maintainer | None |
| docs_url | None |
| author | Odos Matthews |
| requires_python | >=3.8 |
| license | MIT License
Copyright (c) 2021 James Murphy
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. |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
-----------------
# FullmetalAlchemy: Easy-to-use SQL table operations with SQLAlchemy
[](https://pypi.org/project/fullmetalalchemy/)
[](https://pypi.org/project/fullmetalalchemy/)
[](https://github.com/eddiethedean/fullmetalalchemy)
[](https://github.com/eddiethedean/fullmetalalchemy)
[](https://github.com/eddiethedean/fullmetalalchemy)
## What is it?
**FullmetalAlchemy** is a Python package that provides intuitive, high-level functions for common database operations using SQLAlchemy. It simplifies CRUD operations (Create, Read, Update, Delete) while maintaining the power and flexibility of SQLAlchemy under the hood.
### Key Features
- 🔄 **SQLAlchemy 1.4+ and 2.x compatible** - Works seamlessly with both versions
- ⚡ **Async/Await Support** - Full async API for high-performance applications (v2.1.0+)
- 🏗️ **Async Classes** - AsyncTable and AsyncSessionTable for Pythonic async operations (v2.2.0+)
- 📦 **Batch Processing** - Efficient bulk operations with progress tracking and parallel execution (v2.2.0+)
- 🎯 **Simple API** - Intuitive functions for common database operations
- 🔒 **Transaction Management** - Built-in context managers for safe operations
- 📊 **Pythonic Interface** - Array-like access and familiar Python patterns
- 🚀 **Memory Efficient** - Chunked iteration and concurrent processing for large datasets
- 🛡️ **Type Safe** - Full type hints with MyPy strict mode compliance
- ✅ **Thoroughly Tested** - 84% test coverage with 336 passing tests (258 sync + 78 async)
- 🎨 **Code Quality** - Ruff and MyPy strict mode verified
## Installation
```sh
# Install from PyPI (sync operations only)
pip install fullmetalalchemy
# Install with async support
pip install fullmetalalchemy[async]
# Install for development
pip install fullmetalalchemy[dev]
```
The source code is hosted on GitHub at: https://github.com/eddiethedean/fullmetalalchemy
## Dependencies
**Core Dependencies:**
- **SQLAlchemy** (>=1.4, <3) - Python SQL toolkit and ORM
- **tinytim** (>=0.1.2) - Data transformation utilities
- **frozendict** (>=2.4) - Immutable dictionary support
**Optional Async Dependencies** (install with `[async]`):
- **aiosqlite** (>=0.19) - Async SQLite driver
- **greenlet** (>=3.0) - Greenlet concurrency support
- **asyncpg** (>=0.29) - Async PostgreSQL driver (optional)
- **aiomysql** (>=0.2) - Async MySQL driver (optional)
## Quick Start
### Basic CRUD Operations
```python
import fullmetalalchemy as fa
# Create a database connection
engine = fa.create_engine('sqlite:///mydata.db')
# Create a table with some initial data
table = fa.create.create_table_from_records(
'employees',
[
{'id': 1, 'name': 'Alice', 'department': 'Engineering', 'salary': 95000},
{'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 75000}
],
primary_key='id',
engine=engine
)
# Get table for operations
table = fa.get_table('employees', engine)
# SELECT: Get all records
records = fa.select.select_records_all(table, engine)
print(records)
# Output:
# [{'id': 1, 'name': 'Alice', 'department': 'Engineering', 'salary': 95000},
# {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 75000}]
# INSERT: Add new records
fa.insert.insert_records(
table,
[
{'id': 3, 'name': 'Charlie', 'department': 'Engineering', 'salary': 88000},
{'id': 4, 'name': 'Diana', 'department': 'Marketing', 'salary': 82000}
],
engine
)
# Now table has 4 records
# UPDATE: Modify existing records
fa.update.update_records(
table,
[{'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 80000}],
engine
)
record = fa.select.select_record_by_primary_key(table, {'id': 2}, engine)
print(record)
# Output: {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 80000}
# DELETE: Remove records
fa.delete.delete_records(table, 'id', [1, 3], engine)
remaining = fa.select.select_records_all(table, engine)
print(f"Remaining records: {len(remaining)}")
# Output: Remaining records: 2
```
## Usage Examples
### 1. SessionTable - Transaction Management
Use `SessionTable` with context managers for automatic commit/rollback handling:
```python
import fullmetalalchemy as fa
engine = fa.create_engine('sqlite:///products.db')
# Create initial table
fa.create.create_table_from_records(
'products',
[
{'id': 1, 'name': 'Laptop', 'price': 999, 'stock': 10},
{'id': 2, 'name': 'Mouse', 'price': 25, 'stock': 50}
],
primary_key='id',
engine=engine
)
# Use context manager - automatically commits on success, rolls back on error
with fa.SessionTable('products', engine) as table:
# All operations are part of a single transaction
table.insert_records([{'id': 3, 'name': 'Keyboard', 'price': 75, 'stock': 30}])
table.update_records([{'id': 2, 'name': 'Mouse', 'price': 29, 'stock': 45}])
# Automatically commits here if no exceptions
# Verify changes persisted
table = fa.get_table('products', engine)
records = fa.select.select_records_all(table, engine)
print(records)
# Output:
# [{'id': 1, 'name': 'Laptop', 'price': 999, 'stock': 10},
# {'id': 2, 'name': 'Mouse', 'price': 29, 'stock': 45},
# {'id': 3, 'name': 'Keyboard', 'price': 75, 'stock': 30}]
```
### 2. Table Class - Pythonic Interface
The `Table` class provides an intuitive, array-like interface:
```python
import fullmetalalchemy as fa
engine = fa.create_engine('sqlite:///orders.db')
# Create initial table
fa.create.create_table_from_records(
'orders',
[
{'id': 1, 'customer': 'John', 'total': 150.00},
{'id': 2, 'customer': 'Jane', 'total': 200.00}
],
primary_key='id',
engine=engine
)
# Create Table instance
table = fa.Table('orders', engine)
# Access table properties
print(f"Columns: {table.column_names}")
# Output: Columns: ['id', 'customer', 'total']
print(f"Row count: {len(table)}")
# Output: Row count: 2
# Array-like access
print(table[0])
# Output: {'id': 1, 'customer': 'John', 'total': 150.0}
print(table['customer'])
# Output: ['John', 'Jane']
print(table[0:2])
# Output: [{'id': 1, 'customer': 'John', 'total': 150.0},
# {'id': 2, 'customer': 'Jane', 'total': 200.0}]
# Direct operations (auto-commit)
table.insert_records([{'id': 3, 'customer': 'Alice', 'total': 175.00}])
table.delete_records('id', [2])
print(f"After operations: {len(table)} records")
# Output: After operations: 2 records
```
### 3. Advanced Queries
FullmetalAlchemy provides powerful querying capabilities:
```python
import fullmetalalchemy as fa
engine = fa.create_engine('sqlite:///users.db')
# Create test data
fa.create.create_table_from_records(
'users',
[
{'id': i, 'name': f'User{i}', 'age': 20 + i * 5,
'city': ['NYC', 'LA', 'Chicago'][i % 3]}
for i in range(1, 11)
],
primary_key='id',
engine=engine
)
table = fa.get_table('users', engine)
# Select specific columns only
records = fa.select.select_records_all(
table, engine,
include_columns=['id', 'name']
)
print(records[:3])
# Output:
# [{'id': 1, 'name': 'User1'},
# {'id': 2, 'name': 'User2'},
# {'id': 3, 'name': 'User3'}]
# Select by slice (rows 2-5)
records = fa.select.select_records_slice(table, 2, 5, engine)
print(records)
# Output:
# [{'id': 3, 'name': 'User3', 'age': 35, 'city': 'NYC'},
# {'id': 4, 'name': 'User4', 'age': 40, 'city': 'LA'},
# {'id': 5, 'name': 'User5', 'age': 45, 'city': 'Chicago'}]
# Get all values from a specific column
cities = fa.select.select_column_values_all(table, 'city', engine)
print(f"Unique cities: {set(cities)}")
# Output: Unique cities: {'NYC', 'Chicago', 'LA'}
# Memory-efficient chunked iteration for large datasets
for chunk_num, chunk in enumerate(
fa.select.select_records_chunks(table, engine, chunksize=3), 1
):
print(f"Chunk {chunk_num}: {len(chunk)} records")
# Output:
# Chunk 1: 3 records
# Chunk 2: 3 records
# Chunk 3: 3 records
# Chunk 4: 1 records
```
## Async/Await Support (v2.1.0+)
FullmetalAlchemy provides full async/await support for high-performance applications. All CRUD operations have async equivalents in the `async_api` namespace.
### Installation for Async
```sh
# Install with async dependencies
pip install fullmetalalchemy[async]
```
### Basic Async Operations
```python
import asyncio
from fullmetalalchemy import async_api
async def main():
# Create async engine with aiosqlite driver
engine = async_api.create_async_engine('sqlite+aiosqlite:///employees.db')
# Create table with initial data
table = await async_api.create.create_table_from_records(
'employees',
[
{'id': 1, 'name': 'Alice', 'department': 'Engineering', 'salary': 95000},
{'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 75000}
],
primary_key='id',
engine=engine
)
# SELECT: Get all records
records = await async_api.select.select_records_all(table, engine)
print(records)
# Output:
# [{'id': 1, 'name': 'Alice', 'department': 'Engineering', 'salary': 95000},
# {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 75000}]
# INSERT: Add new records
await async_api.insert.insert_records(
table,
[
{'id': 3, 'name': 'Charlie', 'department': 'Engineering', 'salary': 88000},
{'id': 4, 'name': 'Diana', 'department': 'Marketing', 'salary': 82000}
],
engine
)
# Total records: 4
# UPDATE: Modify existing records
await async_api.update.update_records(
table,
[{'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 80000}],
engine
)
record = await async_api.select.select_record_by_primary_key(table, {'id': 2}, engine)
print(record)
# Output: {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 80000}
# DELETE: Remove records
await async_api.delete.delete_records_by_values(table, 'id', [1, 3], engine)
remaining = await async_api.select.select_records_all(table, engine)
print(f"Remaining records: {len(remaining)}")
# Output: Remaining records: 2
await engine.dispose()
# Run the async function
asyncio.run(main())
```
### Concurrent Operations
One of the main benefits of async is the ability to run multiple database operations concurrently:
```python
import asyncio
from fullmetalalchemy import async_api
async def main():
engine = async_api.create_async_engine('sqlite+aiosqlite:///shop.db')
# Create three tables concurrently
tables = await asyncio.gather(
async_api.create.create_table_from_records(
'products',
[{'id': 1, 'name': 'Laptop', 'price': 999}],
primary_key='id',
engine=engine
),
async_api.create.create_table_from_records(
'customers',
[{'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}],
primary_key='id',
engine=engine
),
async_api.create.create_table_from_records(
'orders',
[{'id': 1, 'product_id': 1, 'customer_id': 1}],
primary_key='id',
engine=engine
)
)
print(f"Created {len(tables)} tables concurrently")
# Output: Created 3 tables concurrently
# Query all tables concurrently
results = await asyncio.gather(
async_api.select.select_records_all(tables[0], engine),
async_api.select.select_records_all(tables[1], engine),
async_api.select.select_records_all(tables[2], engine)
)
print("Products:", results[0])
# Output: Products: [{'id': 1, 'name': 'Laptop', 'price': 999}]
print("Customers:", results[1])
# Output: Customers: [{'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}]
print("Orders:", results[2])
# Output: Orders: [{'id': 1, 'product_id': 1, 'customer_id': 1}]
await engine.dispose()
asyncio.run(main())
```
### Advanced Async Queries
All query operations from the sync API are available in async:
```python
import asyncio
from fullmetalalchemy import async_api
async def main():
engine = async_api.create_async_engine('sqlite+aiosqlite:///users.db')
# Create test data
table = await async_api.create.create_table_from_records(
'users',
[{'id': i, 'name': f'User{i}', 'age': 20 + i * 5} for i in range(1, 6)],
primary_key='id',
engine=engine
)
# Select with specific columns
records = await async_api.select.select_records_all(
table, engine, include_columns=['id', 'name']
)
print(records[:3])
# Output: [{'id': 1, 'name': 'User1'}, {'id': 2, 'name': 'User2'}, {'id': 3, 'name': 'User3'}]
# Select by slice
records = await async_api.select.select_records_slice(table, 1, 4, engine)
print(records)
# Output: [{'id': 2, 'name': 'User2', 'age': 30},
# {'id': 3, 'name': 'User3', 'age': 35},
# {'id': 4, 'name': 'User4', 'age': 40}]
# Get column values
ages = await async_api.select.select_column_values_all(table, 'age', engine)
print(ages)
# Output: [25, 30, 35, 40, 45]
await engine.dispose()
asyncio.run(main())
```
### Async Database Drivers
FullmetalAlchemy supports multiple async database drivers:
| Database | Driver | Connection String Example |
|----------|--------|---------------------------|
| **SQLite** | `aiosqlite` | `sqlite+aiosqlite:///path/to/db.db` |
| **PostgreSQL** | `asyncpg` | `postgresql+asyncpg://user:pass@localhost/dbname` |
| **MySQL** | `aiomysql` | `mysql+aiomysql://user:pass@localhost/dbname` |
Install the appropriate driver for your database:
```sh
pip install fullmetalalchemy[async] # Includes aiosqlite
pip install asyncpg # For PostgreSQL
pip install aiomysql # For MySQL
```
## Async Classes (v2.1.0+)
### AsyncTable - Pythonic Async Interface
The `AsyncTable` class provides an intuitive, array-like interface for async operations:
```python
import asyncio
from fullmetalalchemy import async_api
async def main():
engine = async_api.create_async_engine('sqlite+aiosqlite:///products.db')
# Create table with initial data
await async_api.create.create_table_from_records(
'products',
[
{'id': 1, 'name': 'Laptop', 'price': 999, 'stock': 10},
{'id': 2, 'name': 'Mouse', 'price': 25, 'stock': 50}
],
primary_key='id',
engine=engine
)
# Use AsyncTable class
async with async_api.AsyncTable('products', engine) as table:
# Array-like access
first_product = await table[0]
print("First product:", first_product)
# Output: First product: {'id': 1, 'name': 'Laptop', 'price': 999, 'stock': 10}
# Get column values
names = await table['name']
print("Product names:", names)
# Output: Product names: ['Laptop', 'Mouse']
# Insert and update
await table.insert_records([{'id': 3, 'name': 'Keyboard', 'price': 75, 'stock': 30}])
await table.update_records([{'id': 2, 'name': 'Mouse', 'price': 29, 'stock': 45}])
# Get count
count = await table.__len__()
print(f"Total products: {count}")
# Output: Total products: 3
# Async iteration
print("All products:")
async for product in table:
print(f" - {product['name']}: ${product['price']}")
# Output:
# All products:
# - Laptop: $999
# - Mouse: $29
# - Keyboard: $75
await engine.dispose()
asyncio.run(main())
```
### AsyncSessionTable - Transaction Management
The `AsyncSessionTable` class provides transaction-safe async operations with automatic commit/rollback:
```python
import asyncio
from fullmetalalchemy import async_api
async def main():
engine = async_api.create_async_engine('sqlite+aiosqlite:///orders.db')
# Create table
await async_api.create.create_table_from_records(
'orders',
[
{'id': 1, 'customer': 'John', 'total': 150.0},
{'id': 2, 'customer': 'Jane', 'total': 200.0}
],
primary_key='id',
engine=engine
)
# Transaction automatically commits on success
async with async_api.AsyncSessionTable('orders', engine) as table:
await table.insert_records([{'id': 3, 'customer': 'Alice', 'total': 175.0}])
await table.update_records([{'id': 1, 'customer': 'John', 'total': 160.0}])
# Auto-commits here
# Verify changes persisted
async with async_api.AsyncSessionTable('orders', engine) as table:
records = await table.select_all()
print(f"Total orders after transaction: {len(records)}")
# Output: Total orders after transaction: 3
# Transaction rolls back on error
try:
async with async_api.AsyncSessionTable('orders', engine) as table:
await table.insert_records([{'id': 4, 'customer': 'Bob', 'total': 225.0}])
raise ValueError("Simulated error")
except ValueError:
pass
# Verify rollback worked
async with async_api.AsyncSessionTable('orders', engine) as table:
records = await table.select_all()
print(f"Total orders after rollback: {len(records)}")
# Output: Total orders after rollback: 3 (Bob's order was rolled back)
await engine.dispose()
asyncio.run(main())
```
## Batch Operations (v2.2.0+)
### Sync Batch Processing
Process large datasets efficiently with the `BatchProcessor`:
```python
from fullmetalalchemy import BatchProcessor
import fullmetalalchemy as fa
engine = fa.create_engine('sqlite:///data.db')
table = fa.get_table('users', engine)
# Create large dataset
large_dataset = [
{'id': i, 'value': i * 10, 'category': f'cat_{i % 5}'}
for i in range(1, 10001)
]
# Process in batches
processor = BatchProcessor(batch_size=1000, show_progress=False)
result = processor.process_batches(
large_dataset,
lambda batch: fa.insert.insert_records(table, batch, engine)
)
print(f"Processed {result.total_records} records in {result.total_batches} batches")
# Output: Processed 10000 records in 10 batches
```
### Async Batch Processing with Parallelism
The `AsyncBatchProcessor` processes multiple batches concurrently for better performance:
```python
import asyncio
from fullmetalalchemy import async_api
async def main():
engine = async_api.create_async_engine('sqlite+aiosqlite:///data.db')
# Create large dataset
large_dataset = [{'id': i, 'value': i * 5} for i in range(1, 5001)]
# Process batches concurrently (up to 5 at once)
processor = async_api.AsyncBatchProcessor(
batch_size=500,
max_concurrent=5,
show_progress=False
)
async def async_insert_batch(batch):
# This would be your actual insert operation
await asyncio.sleep(0.01) # Simulate I/O
result = await processor.process_batches(large_dataset, async_insert_batch)
print(f"Processed {result.total_records} records in {result.total_batches} batches")
# Output: Processed 5000 records in 10 batches
print(f"Max concurrent: {processor.max_concurrent}")
# Output: Max concurrent: 5
await engine.dispose()
asyncio.run(main())
```
### Batch Error Handling
Handle errors gracefully with the `on_error='continue'` option:
```python
import asyncio
from fullmetalalchemy import async_api
async def main():
records = [{'id': i, 'status': 'pending'} for i in range(10)]
# Continue processing even if some batches fail
processor = async_api.AsyncBatchProcessor(batch_size=3, on_error='continue')
async def flaky_operation(batch):
if batch[0]['id'] == 6:
raise RuntimeError("Simulated error")
result = await processor.process_batches(records, flaky_operation)
print(f"Total batches: {result.total_batches}, Failed: {len(result.failed_batches)}")
# Output: Total batches: 4, Failed: 1
print(f"Successfully processed: {result.total_records - len(result.failed_batches) * 3} records")
# Output: Successfully processed: 7 records
asyncio.run(main())
```
## API Overview
### Connection & Table Access
- `fa.create_engine(url)` - Create SQLAlchemy engine
- `fa.get_table(name, engine)` - Get table object for operations
- `fa.get_table_names(engine)` - List all table names in database
### Create Operations
- `fa.create.create_table()` - Create table from specifications
- `fa.create.create_table_from_records()` - Create table from data
- `fa.create.copy_table()` - Duplicate existing table
### Select Operations
- `fa.select.select_records_all()` - Get all records
- `fa.select.select_records_chunks()` - Iterate records in chunks
- `fa.select.select_records_slice()` - Get records by slice
- `fa.select.select_record_by_primary_key()` - Get single record
- `fa.select.select_column_values_all()` - Get all values from column
### Insert Operations
- `fa.insert.insert_records()` - Insert multiple records
- `fa.insert.insert_from_table()` - Copy records from another table
### Update Operations
- `fa.update.update_records()` - Update existing records
### Delete Operations
- `fa.delete.delete_records()` - Delete by column values
- `fa.delete.delete_records_by_values()` - Delete matching records
- `fa.delete.delete_all_records()` - Clear entire table
### Drop Operations
- `fa.drop.drop_table()` - Remove table from database
## Advanced Features
### Type Safety
FullmetalAlchemy is fully typed with MyPy strict mode compliance:
```python
from typing import List, Dict, Any
import fullmetalalchemy as fa
def process_users(engine: fa.types.SqlConnection) -> List[Dict[str, Any]]:
table = fa.get_table('users', engine)
return fa.select.select_records_all(table, engine)
```
### Transaction Control with SessionTable
```python
with fa.SessionTable('orders', engine) as table:
try:
table.insert_records([...])
table.update_records([...])
# Commits automatically if successful
except Exception as e:
# Automatically rolls back on error
print(f"Transaction failed: {e}")
```
### Bulk Operations
For better performance with large datasets:
```python
# Bulk insert
large_dataset = [{'id': i, 'value': i*2} for i in range(10000)]
fa.insert.insert_records(table, large_dataset, engine)
# Chunked processing
for chunk in fa.select.select_records_chunks(table, engine, chunksize=1000):
process_chunk(chunk)
```
## Compatibility
- **Python**: 3.8, 3.9, 3.10, 3.11, 3.12, 3.13
- **SQLAlchemy**: 1.4+ and 2.x
- **Databases**: SQLite, PostgreSQL, MySQL, and any SQLAlchemy-supported database
## Development
### Running Tests
```sh
# Install development dependencies
pip install -e ".[dev]"
# Run tests with coverage
pytest tests/ --cov=src/fullmetalalchemy --cov-report=term-missing
# Run code quality checks
ruff check src/ tests/
mypy src/fullmetalalchemy
```
### Code Quality
This project maintains high standards:
- **84% Test Coverage** - Comprehensive test suite with 336 tests (258 sync + 78 async)
- **MyPy Strict Mode** - Full type safety enforcement
- **Ruff Verified** - Modern Python code style
- **SQLAlchemy 1.4/2.x Dual Support** - Backwards compatible
- **Async/Await Ready** - Full async API with AsyncTable/AsyncSessionTable classes
- **Batch Operations** - Efficient processing with parallel execution support
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request. For major changes, please open an issue first to discuss what you would like to change.
## License
[MIT License](LICENSE)
## Links
- **Documentation**: https://github.com/eddiethedean/fullmetalalchemy
- **Source Code**: https://github.com/eddiethedean/fullmetalalchemy
- **Issue Tracker**: https://github.com/eddiethedean/fullmetalalchemy/issues
- **PyPI**: https://pypi.org/project/fullmetalalchemy/
## Changelog
See [CHANGELOG.md](CHANGELOG.md) for version history and release notes.
Raw data
{
"_id": null,
"home_page": null,
"name": "fullmetalalchemy",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": null,
"author": "Odos Matthews",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/53/27/77637e75ff9fcd549bc19258cb5af21e40e3b2edee865967f1a75a9f7041/fullmetalalchemy-2.2.3.tar.gz",
"platform": null,
"description": "\n-----------------\n\n# FullmetalAlchemy: Easy-to-use SQL table operations with SQLAlchemy\n\n[](https://pypi.org/project/fullmetalalchemy/)\n\n[](https://pypi.org/project/fullmetalalchemy/)\n[](https://github.com/eddiethedean/fullmetalalchemy)\n[](https://github.com/eddiethedean/fullmetalalchemy)\n[](https://github.com/eddiethedean/fullmetalalchemy)\n\n## What is it?\n\n**FullmetalAlchemy** is a Python package that provides intuitive, high-level functions for common database operations using SQLAlchemy. It simplifies CRUD operations (Create, Read, Update, Delete) while maintaining the power and flexibility of SQLAlchemy under the hood.\n\n### Key Features\n\n- \ud83d\udd04 **SQLAlchemy 1.4+ and 2.x compatible** - Works seamlessly with both versions\n- \u26a1 **Async/Await Support** - Full async API for high-performance applications (v2.1.0+)\n- \ud83c\udfd7\ufe0f **Async Classes** - AsyncTable and AsyncSessionTable for Pythonic async operations (v2.2.0+)\n- \ud83d\udce6 **Batch Processing** - Efficient bulk operations with progress tracking and parallel execution (v2.2.0+)\n- \ud83c\udfaf **Simple API** - Intuitive functions for common database operations\n- \ud83d\udd12 **Transaction Management** - Built-in context managers for safe operations\n- \ud83d\udcca **Pythonic Interface** - Array-like access and familiar Python patterns\n- \ud83d\ude80 **Memory Efficient** - Chunked iteration and concurrent processing for large datasets\n- \ud83d\udee1\ufe0f **Type Safe** - Full type hints with MyPy strict mode compliance\n- \u2705 **Thoroughly Tested** - 84% test coverage with 336 passing tests (258 sync + 78 async)\n- \ud83c\udfa8 **Code Quality** - Ruff and MyPy strict mode verified\n\n## Installation\n\n```sh\n# Install from PyPI (sync operations only)\npip install fullmetalalchemy\n\n# Install with async support\npip install fullmetalalchemy[async]\n\n# Install for development\npip install fullmetalalchemy[dev]\n```\n\nThe source code is hosted on GitHub at: https://github.com/eddiethedean/fullmetalalchemy\n\n## Dependencies\n\n**Core Dependencies:**\n- **SQLAlchemy** (>=1.4, <3) - Python SQL toolkit and ORM\n- **tinytim** (>=0.1.2) - Data transformation utilities\n- **frozendict** (>=2.4) - Immutable dictionary support\n\n**Optional Async Dependencies** (install with `[async]`):\n- **aiosqlite** (>=0.19) - Async SQLite driver\n- **greenlet** (>=3.0) - Greenlet concurrency support\n- **asyncpg** (>=0.29) - Async PostgreSQL driver (optional)\n- **aiomysql** (>=0.2) - Async MySQL driver (optional)\n\n## Quick Start\n\n### Basic CRUD Operations\n\n```python\nimport fullmetalalchemy as fa\n\n# Create a database connection\nengine = fa.create_engine('sqlite:///mydata.db')\n\n# Create a table with some initial data\ntable = fa.create.create_table_from_records(\n 'employees',\n [\n {'id': 1, 'name': 'Alice', 'department': 'Engineering', 'salary': 95000},\n {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 75000}\n ],\n primary_key='id',\n engine=engine\n)\n\n# Get table for operations\ntable = fa.get_table('employees', engine)\n\n# SELECT: Get all records\nrecords = fa.select.select_records_all(table, engine)\nprint(records)\n# Output:\n# [{'id': 1, 'name': 'Alice', 'department': 'Engineering', 'salary': 95000},\n# {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 75000}]\n\n# INSERT: Add new records\nfa.insert.insert_records(\n table,\n [\n {'id': 3, 'name': 'Charlie', 'department': 'Engineering', 'salary': 88000},\n {'id': 4, 'name': 'Diana', 'department': 'Marketing', 'salary': 82000}\n ],\n engine\n)\n# Now table has 4 records\n\n# UPDATE: Modify existing records\nfa.update.update_records(\n table,\n [{'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 80000}],\n engine\n)\nrecord = fa.select.select_record_by_primary_key(table, {'id': 2}, engine)\nprint(record)\n# Output: {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 80000}\n\n# DELETE: Remove records\nfa.delete.delete_records(table, 'id', [1, 3], engine)\nremaining = fa.select.select_records_all(table, engine)\nprint(f\"Remaining records: {len(remaining)}\")\n# Output: Remaining records: 2\n```\n\n## Usage Examples\n\n### 1. SessionTable - Transaction Management\n\nUse `SessionTable` with context managers for automatic commit/rollback handling:\n\n```python\nimport fullmetalalchemy as fa\n\nengine = fa.create_engine('sqlite:///products.db')\n\n# Create initial table\nfa.create.create_table_from_records(\n 'products',\n [\n {'id': 1, 'name': 'Laptop', 'price': 999, 'stock': 10},\n {'id': 2, 'name': 'Mouse', 'price': 25, 'stock': 50}\n ],\n primary_key='id',\n engine=engine\n)\n\n# Use context manager - automatically commits on success, rolls back on error\nwith fa.SessionTable('products', engine) as table:\n # All operations are part of a single transaction\n table.insert_records([{'id': 3, 'name': 'Keyboard', 'price': 75, 'stock': 30}])\n table.update_records([{'id': 2, 'name': 'Mouse', 'price': 29, 'stock': 45}])\n # Automatically commits here if no exceptions\n\n# Verify changes persisted\ntable = fa.get_table('products', engine)\nrecords = fa.select.select_records_all(table, engine)\nprint(records)\n# Output:\n# [{'id': 1, 'name': 'Laptop', 'price': 999, 'stock': 10},\n# {'id': 2, 'name': 'Mouse', 'price': 29, 'stock': 45},\n# {'id': 3, 'name': 'Keyboard', 'price': 75, 'stock': 30}]\n```\n\n### 2. Table Class - Pythonic Interface\n\nThe `Table` class provides an intuitive, array-like interface:\n\n```python\nimport fullmetalalchemy as fa\n\nengine = fa.create_engine('sqlite:///orders.db')\n\n# Create initial table\nfa.create.create_table_from_records(\n 'orders',\n [\n {'id': 1, 'customer': 'John', 'total': 150.00},\n {'id': 2, 'customer': 'Jane', 'total': 200.00}\n ],\n primary_key='id',\n engine=engine\n)\n\n# Create Table instance\ntable = fa.Table('orders', engine)\n\n# Access table properties\nprint(f\"Columns: {table.column_names}\")\n# Output: Columns: ['id', 'customer', 'total']\n\nprint(f\"Row count: {len(table)}\")\n# Output: Row count: 2\n\n# Array-like access\nprint(table[0])\n# Output: {'id': 1, 'customer': 'John', 'total': 150.0}\n\nprint(table['customer'])\n# Output: ['John', 'Jane']\n\nprint(table[0:2])\n# Output: [{'id': 1, 'customer': 'John', 'total': 150.0}, \n# {'id': 2, 'customer': 'Jane', 'total': 200.0}]\n\n# Direct operations (auto-commit)\ntable.insert_records([{'id': 3, 'customer': 'Alice', 'total': 175.00}])\ntable.delete_records('id', [2])\n\nprint(f\"After operations: {len(table)} records\")\n# Output: After operations: 2 records\n```\n\n### 3. Advanced Queries\n\nFullmetalAlchemy provides powerful querying capabilities:\n\n```python\nimport fullmetalalchemy as fa\n\nengine = fa.create_engine('sqlite:///users.db')\n\n# Create test data\nfa.create.create_table_from_records(\n 'users',\n [\n {'id': i, 'name': f'User{i}', 'age': 20 + i * 5, \n 'city': ['NYC', 'LA', 'Chicago'][i % 3]}\n for i in range(1, 11)\n ],\n primary_key='id',\n engine=engine\n)\n\ntable = fa.get_table('users', engine)\n\n# Select specific columns only\nrecords = fa.select.select_records_all(\n table, engine, \n include_columns=['id', 'name']\n)\nprint(records[:3])\n# Output:\n# [{'id': 1, 'name': 'User1'}, \n# {'id': 2, 'name': 'User2'}, \n# {'id': 3, 'name': 'User3'}]\n\n# Select by slice (rows 2-5)\nrecords = fa.select.select_records_slice(table, 2, 5, engine)\nprint(records)\n# Output:\n# [{'id': 3, 'name': 'User3', 'age': 35, 'city': 'NYC'},\n# {'id': 4, 'name': 'User4', 'age': 40, 'city': 'LA'},\n# {'id': 5, 'name': 'User5', 'age': 45, 'city': 'Chicago'}]\n\n# Get all values from a specific column\ncities = fa.select.select_column_values_all(table, 'city', engine)\nprint(f\"Unique cities: {set(cities)}\")\n# Output: Unique cities: {'NYC', 'Chicago', 'LA'}\n\n# Memory-efficient chunked iteration for large datasets\nfor chunk_num, chunk in enumerate(\n fa.select.select_records_chunks(table, engine, chunksize=3), 1\n):\n print(f\"Chunk {chunk_num}: {len(chunk)} records\")\n# Output:\n# Chunk 1: 3 records\n# Chunk 2: 3 records\n# Chunk 3: 3 records\n# Chunk 4: 1 records\n```\n\n## Async/Await Support (v2.1.0+)\n\nFullmetalAlchemy provides full async/await support for high-performance applications. All CRUD operations have async equivalents in the `async_api` namespace.\n\n### Installation for Async\n\n```sh\n# Install with async dependencies\npip install fullmetalalchemy[async]\n```\n\n### Basic Async Operations\n\n```python\nimport asyncio\nfrom fullmetalalchemy import async_api\n\nasync def main():\n # Create async engine with aiosqlite driver\n engine = async_api.create_async_engine('sqlite+aiosqlite:///employees.db')\n \n # Create table with initial data\n table = await async_api.create.create_table_from_records(\n 'employees',\n [\n {'id': 1, 'name': 'Alice', 'department': 'Engineering', 'salary': 95000},\n {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 75000}\n ],\n primary_key='id',\n engine=engine\n )\n \n # SELECT: Get all records\n records = await async_api.select.select_records_all(table, engine)\n print(records)\n # Output:\n # [{'id': 1, 'name': 'Alice', 'department': 'Engineering', 'salary': 95000},\n # {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 75000}]\n \n # INSERT: Add new records\n await async_api.insert.insert_records(\n table,\n [\n {'id': 3, 'name': 'Charlie', 'department': 'Engineering', 'salary': 88000},\n {'id': 4, 'name': 'Diana', 'department': 'Marketing', 'salary': 82000}\n ],\n engine\n )\n # Total records: 4\n \n # UPDATE: Modify existing records\n await async_api.update.update_records(\n table,\n [{'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 80000}],\n engine\n )\n record = await async_api.select.select_record_by_primary_key(table, {'id': 2}, engine)\n print(record)\n # Output: {'id': 2, 'name': 'Bob', 'department': 'Sales', 'salary': 80000}\n \n # DELETE: Remove records\n await async_api.delete.delete_records_by_values(table, 'id', [1, 3], engine)\n remaining = await async_api.select.select_records_all(table, engine)\n print(f\"Remaining records: {len(remaining)}\")\n # Output: Remaining records: 2\n \n await engine.dispose()\n\n# Run the async function\nasyncio.run(main())\n```\n\n### Concurrent Operations\n\nOne of the main benefits of async is the ability to run multiple database operations concurrently:\n\n```python\nimport asyncio\nfrom fullmetalalchemy import async_api\n\nasync def main():\n engine = async_api.create_async_engine('sqlite+aiosqlite:///shop.db')\n \n # Create three tables concurrently\n tables = await asyncio.gather(\n async_api.create.create_table_from_records(\n 'products',\n [{'id': 1, 'name': 'Laptop', 'price': 999}],\n primary_key='id',\n engine=engine\n ),\n async_api.create.create_table_from_records(\n 'customers',\n [{'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}],\n primary_key='id',\n engine=engine\n ),\n async_api.create.create_table_from_records(\n 'orders',\n [{'id': 1, 'product_id': 1, 'customer_id': 1}],\n primary_key='id',\n engine=engine\n )\n )\n print(f\"Created {len(tables)} tables concurrently\")\n # Output: Created 3 tables concurrently\n \n # Query all tables concurrently\n results = await asyncio.gather(\n async_api.select.select_records_all(tables[0], engine),\n async_api.select.select_records_all(tables[1], engine),\n async_api.select.select_records_all(tables[2], engine)\n )\n \n print(\"Products:\", results[0])\n # Output: Products: [{'id': 1, 'name': 'Laptop', 'price': 999}]\n print(\"Customers:\", results[1])\n # Output: Customers: [{'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}]\n print(\"Orders:\", results[2])\n # Output: Orders: [{'id': 1, 'product_id': 1, 'customer_id': 1}]\n \n await engine.dispose()\n\nasyncio.run(main())\n```\n\n### Advanced Async Queries\n\nAll query operations from the sync API are available in async:\n\n```python\nimport asyncio\nfrom fullmetalalchemy import async_api\n\nasync def main():\n engine = async_api.create_async_engine('sqlite+aiosqlite:///users.db')\n \n # Create test data\n table = await async_api.create.create_table_from_records(\n 'users',\n [{'id': i, 'name': f'User{i}', 'age': 20 + i * 5} for i in range(1, 6)],\n primary_key='id',\n engine=engine\n )\n \n # Select with specific columns\n records = await async_api.select.select_records_all(\n table, engine, include_columns=['id', 'name']\n )\n print(records[:3])\n # Output: [{'id': 1, 'name': 'User1'}, {'id': 2, 'name': 'User2'}, {'id': 3, 'name': 'User3'}]\n \n # Select by slice\n records = await async_api.select.select_records_slice(table, 1, 4, engine)\n print(records)\n # Output: [{'id': 2, 'name': 'User2', 'age': 30},\n # {'id': 3, 'name': 'User3', 'age': 35},\n # {'id': 4, 'name': 'User4', 'age': 40}]\n \n # Get column values\n ages = await async_api.select.select_column_values_all(table, 'age', engine)\n print(ages)\n # Output: [25, 30, 35, 40, 45]\n \n await engine.dispose()\n\nasyncio.run(main())\n```\n\n### Async Database Drivers\n\nFullmetalAlchemy supports multiple async database drivers:\n\n| Database | Driver | Connection String Example |\n|----------|--------|---------------------------|\n| **SQLite** | `aiosqlite` | `sqlite+aiosqlite:///path/to/db.db` |\n| **PostgreSQL** | `asyncpg` | `postgresql+asyncpg://user:pass@localhost/dbname` |\n| **MySQL** | `aiomysql` | `mysql+aiomysql://user:pass@localhost/dbname` |\n\nInstall the appropriate driver for your database:\n\n```sh\npip install fullmetalalchemy[async] # Includes aiosqlite\npip install asyncpg # For PostgreSQL\npip install aiomysql # For MySQL\n```\n\n## Async Classes (v2.1.0+)\n\n### AsyncTable - Pythonic Async Interface\n\nThe `AsyncTable` class provides an intuitive, array-like interface for async operations:\n\n```python\nimport asyncio\nfrom fullmetalalchemy import async_api\n\nasync def main():\n engine = async_api.create_async_engine('sqlite+aiosqlite:///products.db')\n \n # Create table with initial data\n await async_api.create.create_table_from_records(\n 'products',\n [\n {'id': 1, 'name': 'Laptop', 'price': 999, 'stock': 10},\n {'id': 2, 'name': 'Mouse', 'price': 25, 'stock': 50}\n ],\n primary_key='id',\n engine=engine\n )\n \n # Use AsyncTable class\n async with async_api.AsyncTable('products', engine) as table:\n # Array-like access\n first_product = await table[0]\n print(\"First product:\", first_product)\n # Output: First product: {'id': 1, 'name': 'Laptop', 'price': 999, 'stock': 10}\n \n # Get column values\n names = await table['name']\n print(\"Product names:\", names)\n # Output: Product names: ['Laptop', 'Mouse']\n \n # Insert and update\n await table.insert_records([{'id': 3, 'name': 'Keyboard', 'price': 75, 'stock': 30}])\n await table.update_records([{'id': 2, 'name': 'Mouse', 'price': 29, 'stock': 45}])\n \n # Get count\n count = await table.__len__()\n print(f\"Total products: {count}\")\n # Output: Total products: 3\n \n # Async iteration\n print(\"All products:\")\n async for product in table:\n print(f\" - {product['name']}: ${product['price']}\")\n # Output:\n # All products:\n # - Laptop: $999\n # - Mouse: $29\n # - Keyboard: $75\n \n await engine.dispose()\n\nasyncio.run(main())\n```\n\n### AsyncSessionTable - Transaction Management\n\nThe `AsyncSessionTable` class provides transaction-safe async operations with automatic commit/rollback:\n\n```python\nimport asyncio\nfrom fullmetalalchemy import async_api\n\nasync def main():\n engine = async_api.create_async_engine('sqlite+aiosqlite:///orders.db')\n \n # Create table\n await async_api.create.create_table_from_records(\n 'orders',\n [\n {'id': 1, 'customer': 'John', 'total': 150.0},\n {'id': 2, 'customer': 'Jane', 'total': 200.0}\n ],\n primary_key='id',\n engine=engine\n )\n \n # Transaction automatically commits on success\n async with async_api.AsyncSessionTable('orders', engine) as table:\n await table.insert_records([{'id': 3, 'customer': 'Alice', 'total': 175.0}])\n await table.update_records([{'id': 1, 'customer': 'John', 'total': 160.0}])\n # Auto-commits here\n \n # Verify changes persisted\n async with async_api.AsyncSessionTable('orders', engine) as table:\n records = await table.select_all()\n print(f\"Total orders after transaction: {len(records)}\")\n # Output: Total orders after transaction: 3\n \n # Transaction rolls back on error\n try:\n async with async_api.AsyncSessionTable('orders', engine) as table:\n await table.insert_records([{'id': 4, 'customer': 'Bob', 'total': 225.0}])\n raise ValueError(\"Simulated error\")\n except ValueError:\n pass\n \n # Verify rollback worked\n async with async_api.AsyncSessionTable('orders', engine) as table:\n records = await table.select_all()\n print(f\"Total orders after rollback: {len(records)}\")\n # Output: Total orders after rollback: 3 (Bob's order was rolled back)\n \n await engine.dispose()\n\nasyncio.run(main())\n```\n\n## Batch Operations (v2.2.0+)\n\n### Sync Batch Processing\n\nProcess large datasets efficiently with the `BatchProcessor`:\n\n```python\nfrom fullmetalalchemy import BatchProcessor\nimport fullmetalalchemy as fa\n\nengine = fa.create_engine('sqlite:///data.db')\ntable = fa.get_table('users', engine)\n\n# Create large dataset\nlarge_dataset = [\n {'id': i, 'value': i * 10, 'category': f'cat_{i % 5}'}\n for i in range(1, 10001)\n]\n\n# Process in batches\nprocessor = BatchProcessor(batch_size=1000, show_progress=False)\n\nresult = processor.process_batches(\n large_dataset,\n lambda batch: fa.insert.insert_records(table, batch, engine)\n)\n\nprint(f\"Processed {result.total_records} records in {result.total_batches} batches\")\n# Output: Processed 10000 records in 10 batches\n```\n\n### Async Batch Processing with Parallelism\n\nThe `AsyncBatchProcessor` processes multiple batches concurrently for better performance:\n\n```python\nimport asyncio\nfrom fullmetalalchemy import async_api\n\nasync def main():\n engine = async_api.create_async_engine('sqlite+aiosqlite:///data.db')\n \n # Create large dataset\n large_dataset = [{'id': i, 'value': i * 5} for i in range(1, 5001)]\n \n # Process batches concurrently (up to 5 at once)\n processor = async_api.AsyncBatchProcessor(\n batch_size=500,\n max_concurrent=5,\n show_progress=False\n )\n \n async def async_insert_batch(batch):\n # This would be your actual insert operation\n await asyncio.sleep(0.01) # Simulate I/O\n \n result = await processor.process_batches(large_dataset, async_insert_batch)\n \n print(f\"Processed {result.total_records} records in {result.total_batches} batches\")\n # Output: Processed 5000 records in 10 batches\n print(f\"Max concurrent: {processor.max_concurrent}\")\n # Output: Max concurrent: 5\n \n await engine.dispose()\n\nasyncio.run(main())\n```\n\n### Batch Error Handling\n\nHandle errors gracefully with the `on_error='continue'` option:\n\n```python\nimport asyncio\nfrom fullmetalalchemy import async_api\n\nasync def main():\n records = [{'id': i, 'status': 'pending'} for i in range(10)]\n \n # Continue processing even if some batches fail\n processor = async_api.AsyncBatchProcessor(batch_size=3, on_error='continue')\n \n async def flaky_operation(batch):\n if batch[0]['id'] == 6:\n raise RuntimeError(\"Simulated error\")\n \n result = await processor.process_batches(records, flaky_operation)\n \n print(f\"Total batches: {result.total_batches}, Failed: {len(result.failed_batches)}\")\n # Output: Total batches: 4, Failed: 1\n print(f\"Successfully processed: {result.total_records - len(result.failed_batches) * 3} records\")\n # Output: Successfully processed: 7 records\n\nasyncio.run(main())\n```\n\n## API Overview\n\n### Connection & Table Access\n- `fa.create_engine(url)` - Create SQLAlchemy engine\n- `fa.get_table(name, engine)` - Get table object for operations\n- `fa.get_table_names(engine)` - List all table names in database\n\n### Create Operations\n- `fa.create.create_table()` - Create table from specifications\n- `fa.create.create_table_from_records()` - Create table from data\n- `fa.create.copy_table()` - Duplicate existing table\n\n### Select Operations\n- `fa.select.select_records_all()` - Get all records\n- `fa.select.select_records_chunks()` - Iterate records in chunks\n- `fa.select.select_records_slice()` - Get records by slice\n- `fa.select.select_record_by_primary_key()` - Get single record\n- `fa.select.select_column_values_all()` - Get all values from column\n\n### Insert Operations\n- `fa.insert.insert_records()` - Insert multiple records\n- `fa.insert.insert_from_table()` - Copy records from another table\n\n### Update Operations\n- `fa.update.update_records()` - Update existing records\n\n### Delete Operations\n- `fa.delete.delete_records()` - Delete by column values\n- `fa.delete.delete_records_by_values()` - Delete matching records\n- `fa.delete.delete_all_records()` - Clear entire table\n\n### Drop Operations\n- `fa.drop.drop_table()` - Remove table from database\n\n## Advanced Features\n\n### Type Safety\n\nFullmetalAlchemy is fully typed with MyPy strict mode compliance:\n\n```python\nfrom typing import List, Dict, Any\nimport fullmetalalchemy as fa\n\ndef process_users(engine: fa.types.SqlConnection) -> List[Dict[str, Any]]:\n table = fa.get_table('users', engine)\n return fa.select.select_records_all(table, engine)\n```\n\n### Transaction Control with SessionTable\n\n```python\nwith fa.SessionTable('orders', engine) as table:\n try:\n table.insert_records([...])\n table.update_records([...])\n # Commits automatically if successful\n except Exception as e:\n # Automatically rolls back on error\n print(f\"Transaction failed: {e}\")\n```\n\n### Bulk Operations\n\nFor better performance with large datasets:\n\n```python\n# Bulk insert\nlarge_dataset = [{'id': i, 'value': i*2} for i in range(10000)]\nfa.insert.insert_records(table, large_dataset, engine)\n\n# Chunked processing\nfor chunk in fa.select.select_records_chunks(table, engine, chunksize=1000):\n process_chunk(chunk)\n```\n\n## Compatibility\n\n- **Python**: 3.8, 3.9, 3.10, 3.11, 3.12, 3.13\n- **SQLAlchemy**: 1.4+ and 2.x\n- **Databases**: SQLite, PostgreSQL, MySQL, and any SQLAlchemy-supported database\n\n## Development\n\n### Running Tests\n\n```sh\n# Install development dependencies\npip install -e \".[dev]\"\n\n# Run tests with coverage\npytest tests/ --cov=src/fullmetalalchemy --cov-report=term-missing\n\n# Run code quality checks\nruff check src/ tests/\nmypy src/fullmetalalchemy\n```\n\n### Code Quality\n\nThis project maintains high standards:\n- **84% Test Coverage** - Comprehensive test suite with 336 tests (258 sync + 78 async)\n- **MyPy Strict Mode** - Full type safety enforcement\n- **Ruff Verified** - Modern Python code style\n- **SQLAlchemy 1.4/2.x Dual Support** - Backwards compatible\n- **Async/Await Ready** - Full async API with AsyncTable/AsyncSessionTable classes\n- **Batch Operations** - Efficient processing with parallel execution support\n\n## Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request. For major changes, please open an issue first to discuss what you would like to change.\n\n## License\n\n[MIT License](LICENSE)\n\n## Links\n\n- **Documentation**: https://github.com/eddiethedean/fullmetalalchemy\n- **Source Code**: https://github.com/eddiethedean/fullmetalalchemy\n- **Issue Tracker**: https://github.com/eddiethedean/fullmetalalchemy/issues\n- **PyPI**: https://pypi.org/project/fullmetalalchemy/\n\n## Changelog\n\nSee [CHANGELOG.md](CHANGELOG.md) for version history and release notes.\n",
"bugtrack_url": null,
"license": "MIT License\n \n Copyright (c) 2021 James Murphy\n \n Permission is hereby granted, free of charge, to any person obtaining a copy\n of this software and associated documentation files (the \"Software\"), to deal\n in the Software without restriction, including without limitation the rights\n to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n copies of the Software, and to permit persons to whom the Software is\n furnished to do so, subject to the following conditions:\n \n The above copyright notice and this permission notice shall be included in all\n copies or substantial portions of the Software.\n \n THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\n SOFTWARE.",
"summary": "Easy-to-use helpers for SQL table changes with SQLAlchemy.",
"version": "2.2.3",
"project_urls": {
"Homepage": "https://github.com/eddiethedean/fullmetalalchemy",
"Repository": "https://github.com/eddiethedean/fullmetalalchemy"
},
"split_keywords": [],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "dda8d3cb391868bbfff2c7a07381112a2bfbd8cf25af8ea0dcbfe29fd8fb2484",
"md5": "a2cc19273c1883d861472310d3f41442",
"sha256": "6316b40dc53f14cd2de45c21cc7359f274d9e9b09d1c3f85e7542868ba19ff54"
},
"downloads": -1,
"filename": "fullmetalalchemy-2.2.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a2cc19273c1883d861472310d3f41442",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 59544,
"upload_time": "2025-10-18T19:13:13",
"upload_time_iso_8601": "2025-10-18T19:13:13.353672Z",
"url": "https://files.pythonhosted.org/packages/dd/a8/d3cb391868bbfff2c7a07381112a2bfbd8cf25af8ea0dcbfe29fd8fb2484/fullmetalalchemy-2.2.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "532777637e75ff9fcd549bc19258cb5af21e40e3b2edee865967f1a75a9f7041",
"md5": "58163ab4b98e2839dbd1746e28d1b796",
"sha256": "9b9f5498799d2d3953b32d6f41d42f7c97b7c8ca13db2f2d252dad21ab59ad47"
},
"downloads": -1,
"filename": "fullmetalalchemy-2.2.3.tar.gz",
"has_sig": false,
"md5_digest": "58163ab4b98e2839dbd1746e28d1b796",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 67886,
"upload_time": "2025-10-18T19:13:15",
"upload_time_iso_8601": "2025-10-18T19:13:15.278174Z",
"url": "https://files.pythonhosted.org/packages/53/27/77637e75ff9fcd549bc19258cb5af21e40e3b2edee865967f1a75a9f7041/fullmetalalchemy-2.2.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-18 19:13:15",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "eddiethedean",
"github_project": "fullmetalalchemy",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "fullmetalalchemy"
}
