# SQLAlchemy Tenant Wiper
A flexible SQLAlchemy-based library for tenant data deletion in multi-tenant applications. Supports custom dynamic tenant column filtering using lambda expressions, robust validations and handles complex relationship paths for tables without explicit tenant knowledge
[](https://badge.fury.io/py/sqlalchemy-tenant-wiper)
[](https://pypi.org/project/sqlalchemy-tenant-wiper/)
[](https://opensource.org/licenses/MIT)
## Features
- **Dynamic Tenant Filtering**: Lambda expressions adapt to each table's columns
- **Relationship Path Support**: Declare relationships for tables that indirectly belong to a tenants (See below)
- **Composite Primary Key Support**: Properly handles tables with composite primary keys
- **Configuration-Time Validation**: Comprehensive runtime validation at table & column level
- **Two-Phase Deletion**: Safe deletion order respecting foreign key constraints
- **Dry Run Mode**: Preview what would be deleted before execution
- **Batched Deletions**: Efficient deletion in configurable batch sizes
## Installation
### Using pip
```bash
pip install sqlalchemy-tenant-wiper
```
### Using Poetry
```bash
poetry add sqlalchemy-tenant-wiper
```
## Quick Start
```python
from sqlalchemy_tenant_wiper import TenantWiperConfig, TenantDeleter
from uuid import UUID
# Configure your tenant deletion settings
config = TenantWiperConfig(
base=YourSQLAlchemyBase,
tenant_filters=[
# A record is targeted for deletion if it matches *any* of these simple conditions you provide.
lambda table: table.c.tenant_id == your_tenant_uuid,
lambda table: table.c.org_id.in_(your_org_uuids)
],
tenant_join_paths=[
# tables with 'tenant_id' or 'org_id' column doesn't require declaration
# and explicitly added for removal
# <table_name>__<from_key>=<to_key>__<to_table_name>...
'products__id=product_id__order_items__order_id=id__orders'
],
excluded_tables=['audit_logs', 'system_logs'],
batch_size=500
)
# Create deleter and execute
deleter = TenantDeleter(config)
deleter.delete(session, dry_run=True) # Preview first
deleter.delete(session, commit=True) # Execute deletion
```
## Usage Examples
### Basic Tenant Filtering
```python
from sqlalchemy_tenant_wiper import TenantWiperConfig, TenantDeleter
from uuid import UUID
# Simple tenant filtering by tenant_id
tenant_uuid = UUID('12345678-1234-5678-9012-123456789abc')
config = TenantWiperConfig(
base=Base,
tenant_filters=[
lambda table: table.c.tenant_id == str(tenant_uuid)
]
)
deleter = TenantDeleter(config)
deleter.delete(session, dry_run=True, commit=False)
```
### Multiple Tenant Filters
```python
# Multiple ways to identify tenant data
config = TenantWiperConfig(
base=Base,
tenant_filters=[
lambda table: table.c.tenant_id == str(tenant_uuid),
lambda table: table.c.organization_id.in_(org_uuids),
lambda table: table.c.customer_code == 'ACME_CORP'
]
)
```
### Relationship Paths for Indirect Tables
```python
# Handle tables without direct tenant columns
config = TenantWiperConfig(
base=Base,
tenant_filters=[
lambda table: table.c.tenant_id == str(tenant_uuid)
],
tenant_join_paths=[
# <table_name>__<from_key>=<to_key>__<to_table_name>...
# Products -> OrderItems -> Orders (with tenant_id)
'products__id=product_id__order_items__order_id=id__orders',
# Audit logs -> Users (with tenant_id)
'audit_logs__user_id=id__users',
# Complex multi-hop relationship
'categories__id=category_id__products__id=product_id__order_items__order_id=id__orders'
]
)
```
### Configuration Validation
```python
# Validate configuration at startup/in tests
config = TenantWiperConfig(
base=Base,
tenant_filters=[...],
tenant_join_paths=[...],
validate_on_init=True # Default: validates on creation
)
# Or validate explicitly
config = TenantWiperConfig(base=Base, validate_on_init=False)
config.validate() # Call when ready
```
### Excluding Tables
```python
# Skip certain tables from deletion
config = TenantWiperConfig(
base=Base,
tenant_filters=[...],
excluded_tables=[
'audit_logs',
'system_configs',
'migration_history'
]
)
```
## Configuration Reference
### TenantWiperConfig
| Parameter | Type | Description |
|-----------|------|-------------|
| `base` | SQLAlchemy Base | Your declarative base class |
| `tenant_filters` | `List[Callable[[Table], Any]]` | Lambda functions for tenant filtering |
| `tenant_join_paths` | `List[str]` | Relationship path strings for indirect tables |
| `excluded_tables` | `List[str]` | Table names to exclude from deletion |
| `validate_on_init` | `bool` | Whether to validate config on creation (default: True) |
### TenantDeleter.delete()
| Parameter | Type | Description |
|-----------|------|-------------|
| `session` | SQLAlchemy Session | Database session for operations |
| `dry_run` | `bool` | If True, only report what would be deleted |
| `commit` | `bool` | If True, commit the transaction |
## Relationship Path Syntax
Relationship paths use double underscore (`__`) separators:
```
table1__from_key=to_key__table2__from_key2=to_key2__table3
```
Example:
```
products__id=product_id__order_items__order_id=id__orders
```
This creates joins:
1. `products.id = order_items.product_id`
2. `order_items.order_id = orders.id`
The final table (`orders`) must have tenant filter columns.
## Requirements
- Python 3.8+
- SQLAlchemy 1.4+
## Development
### Setup
```bash
git clone https://github.com/yourusername/sqlalchemy-tenant-wiper.git
cd sqlalchemy-tenant-wiper
poetry install
```
### Testing
```bash
poetry run pytest
```
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## TODO
[ ] Make enforcing listing of all tables as optional
Raw data
{
"_id": null,
"home_page": null,
"name": "sqlalchemy-tenant-wiper",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": null,
"keywords": "sqlalchemy, tenant, multi-tenant, database, deletion, cleanup",
"author": "Your Name",
"author_email": "your.email@example.com",
"download_url": "https://files.pythonhosted.org/packages/69/38/6ee59f97693c6ecd2bea0dc90b839c1a02829e9c3116ef5ff1ca1cf26fe6/sqlalchemy_tenant_wiper-0.1.8.tar.gz",
"platform": null,
"description": "# SQLAlchemy Tenant Wiper\n\nA flexible SQLAlchemy-based library for tenant data deletion in multi-tenant applications. Supports custom dynamic tenant column filtering using lambda expressions, robust validations and handles complex relationship paths for tables without explicit tenant knowledge\n\n[](https://badge.fury.io/py/sqlalchemy-tenant-wiper)\n[](https://pypi.org/project/sqlalchemy-tenant-wiper/)\n[](https://opensource.org/licenses/MIT)\n\n## Features\n\n- **Dynamic Tenant Filtering**: Lambda expressions adapt to each table's columns\n- **Relationship Path Support**: Declare relationships for tables that indirectly belong to a tenants (See below)\n- **Composite Primary Key Support**: Properly handles tables with composite primary keys\n- **Configuration-Time Validation**: Comprehensive runtime validation at table & column level\n- **Two-Phase Deletion**: Safe deletion order respecting foreign key constraints\n- **Dry Run Mode**: Preview what would be deleted before execution\n- **Batched Deletions**: Efficient deletion in configurable batch sizes\n\n## Installation\n\n### Using pip\n\n```bash\npip install sqlalchemy-tenant-wiper\n```\n\n### Using Poetry\n\n```bash\npoetry add sqlalchemy-tenant-wiper\n```\n\n## Quick Start\n\n```python\nfrom sqlalchemy_tenant_wiper import TenantWiperConfig, TenantDeleter\nfrom uuid import UUID\n\n# Configure your tenant deletion settings\nconfig = TenantWiperConfig(\n base=YourSQLAlchemyBase,\n tenant_filters=[\n # A record is targeted for deletion if it matches *any* of these simple conditions you provide.\n lambda table: table.c.tenant_id == your_tenant_uuid,\n lambda table: table.c.org_id.in_(your_org_uuids)\n ],\n tenant_join_paths=[\n # tables with 'tenant_id' or 'org_id' column doesn't require declaration \n # and explicitly added for removal\n\n # <table_name>__<from_key>=<to_key>__<to_table_name>...\n 'products__id=product_id__order_items__order_id=id__orders'\n ],\n excluded_tables=['audit_logs', 'system_logs'],\n batch_size=500\n)\n\n# Create deleter and execute\ndeleter = TenantDeleter(config)\ndeleter.delete(session, dry_run=True) # Preview first\ndeleter.delete(session, commit=True) # Execute deletion\n```\n\n## Usage Examples\n\n### Basic Tenant Filtering\n\n```python\nfrom sqlalchemy_tenant_wiper import TenantWiperConfig, TenantDeleter\nfrom uuid import UUID\n\n# Simple tenant filtering by tenant_id\ntenant_uuid = UUID('12345678-1234-5678-9012-123456789abc')\n\nconfig = TenantWiperConfig(\n base=Base,\n tenant_filters=[\n lambda table: table.c.tenant_id == str(tenant_uuid)\n ]\n)\n\ndeleter = TenantDeleter(config)\ndeleter.delete(session, dry_run=True, commit=False)\n```\n\n### Multiple Tenant Filters\n\n```python\n# Multiple ways to identify tenant data\nconfig = TenantWiperConfig(\n base=Base,\n tenant_filters=[\n lambda table: table.c.tenant_id == str(tenant_uuid),\n lambda table: table.c.organization_id.in_(org_uuids),\n lambda table: table.c.customer_code == 'ACME_CORP'\n ]\n)\n```\n\n### Relationship Paths for Indirect Tables\n\n```python\n# Handle tables without direct tenant columns\nconfig = TenantWiperConfig(\n base=Base,\n tenant_filters=[\n lambda table: table.c.tenant_id == str(tenant_uuid)\n ],\n tenant_join_paths=[\n # <table_name>__<from_key>=<to_key>__<to_table_name>...\n # Products -> OrderItems -> Orders (with tenant_id)\n 'products__id=product_id__order_items__order_id=id__orders',\n \n # Audit logs -> Users (with tenant_id) \n 'audit_logs__user_id=id__users',\n \n # Complex multi-hop relationship\n 'categories__id=category_id__products__id=product_id__order_items__order_id=id__orders'\n ]\n)\n```\n\n### Configuration Validation\n\n```python\n# Validate configuration at startup/in tests\nconfig = TenantWiperConfig(\n base=Base,\n tenant_filters=[...],\n tenant_join_paths=[...],\n validate_on_init=True # Default: validates on creation\n)\n\n# Or validate explicitly\nconfig = TenantWiperConfig(base=Base, validate_on_init=False)\nconfig.validate() # Call when ready\n```\n\n### Excluding Tables\n\n```python\n# Skip certain tables from deletion\nconfig = TenantWiperConfig(\n base=Base,\n tenant_filters=[...],\n excluded_tables=[\n 'audit_logs',\n 'system_configs', \n 'migration_history'\n ]\n)\n```\n\n## Configuration Reference\n\n### TenantWiperConfig\n\n| Parameter | Type | Description |\n|-----------|------|-------------|\n| `base` | SQLAlchemy Base | Your declarative base class |\n| `tenant_filters` | `List[Callable[[Table], Any]]` | Lambda functions for tenant filtering |\n| `tenant_join_paths` | `List[str]` | Relationship path strings for indirect tables |\n| `excluded_tables` | `List[str]` | Table names to exclude from deletion |\n| `validate_on_init` | `bool` | Whether to validate config on creation (default: True) |\n\n### TenantDeleter.delete()\n\n| Parameter | Type | Description |\n|-----------|------|-------------|\n| `session` | SQLAlchemy Session | Database session for operations |\n| `dry_run` | `bool` | If True, only report what would be deleted |\n| `commit` | `bool` | If True, commit the transaction |\n\n## Relationship Path Syntax\n\nRelationship paths use double underscore (`__`) separators:\n\n```\ntable1__from_key=to_key__table2__from_key2=to_key2__table3\n```\n\nExample:\n```\nproducts__id=product_id__order_items__order_id=id__orders\n```\n\nThis creates joins:\n1. `products.id = order_items.product_id`\n2. `order_items.order_id = orders.id`\n\nThe final table (`orders`) must have tenant filter columns.\n\n## Requirements\n\n- Python 3.8+\n- SQLAlchemy 1.4+\n\n## Development\n\n### Setup\n\n```bash\ngit clone https://github.com/yourusername/sqlalchemy-tenant-wiper.git\ncd sqlalchemy-tenant-wiper\npoetry install\n```\n\n### Testing\n\n```bash\npoetry run pytest\n```\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n\n## TODO\n[ ] Make enforcing listing of all tables as optional \n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A flexible SQLAlchemy-based library for tenant data deletion in multi-tenant applications",
"version": "0.1.8",
"project_urls": {
"Documentation": "https://github.com/timabilov/sqlalchemy-tenant-wiper#readme",
"Homepage": "https://github.com/timabilov/sqlalchemy-tenant-wiper",
"Repository": "https://github.com/timabilov/sqlalchemy-tenant-wiper"
},
"split_keywords": [
"sqlalchemy",
" tenant",
" multi-tenant",
" database",
" deletion",
" cleanup"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "6b7219a94d8bd32c728019ae42675123a43aafcdf10eda568aedff4a729dec30",
"md5": "3ec790eba9c88b49e07cea4260c4cafd",
"sha256": "70754c5ab8f3a629920bff4d030f49eb3ec1068c208c377aea75c08ded7dd723"
},
"downloads": -1,
"filename": "sqlalchemy_tenant_wiper-0.1.8-py3-none-any.whl",
"has_sig": false,
"md5_digest": "3ec790eba9c88b49e07cea4260c4cafd",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 11054,
"upload_time": "2025-07-14T17:51:33",
"upload_time_iso_8601": "2025-07-14T17:51:33.432079Z",
"url": "https://files.pythonhosted.org/packages/6b/72/19a94d8bd32c728019ae42675123a43aafcdf10eda568aedff4a729dec30/sqlalchemy_tenant_wiper-0.1.8-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "69386ee59f97693c6ecd2bea0dc90b839c1a02829e9c3116ef5ff1ca1cf26fe6",
"md5": "b63892e89e3f49fc47f9fead432308de",
"sha256": "ea6729797d3834fb4c35ff7288a8166be60bf237930af0c6ef5727bf1af22fbc"
},
"downloads": -1,
"filename": "sqlalchemy_tenant_wiper-0.1.8.tar.gz",
"has_sig": false,
"md5_digest": "b63892e89e3f49fc47f9fead432308de",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 12405,
"upload_time": "2025-07-14T17:51:34",
"upload_time_iso_8601": "2025-07-14T17:51:34.687069Z",
"url": "https://files.pythonhosted.org/packages/69/38/6ee59f97693c6ecd2bea0dc90b839c1a02829e9c3116ef5ff1ca1cf26fe6/sqlalchemy_tenant_wiper-0.1.8.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-14 17:51:34",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "timabilov",
"github_project": "sqlalchemy-tenant-wiper#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "sqlalchemy-tenant-wiper"
}
JSON
