anonimize 0.1.0

A flexible, backend-agnostic library for data anonymization (Pandas, Polars, Streams).

Anonimize Library

anonimize is a flexible, agnostic library designed for data anonymization. It decouples anonymization logic from data processing, allowing it to work seamlessly with Pandas, Polars, or pure Python Streams/Iterators.

Installation

The core library is lightweight (only faker and numpy). You can install it with specific backends depending on your needs:

# Minimum install (Streams & Cursors only)
pip install anonimize

# With Pandas support
pip install "anonimize[pandas]"

# With Polars support
pip install "anonimize[polars]"

# Install everything
pip install "anonimize[all]"

Core Concepts

The library is built on a Strategy Pattern architecture:

1. Configuration (Schema): A dictionary defining {column_name: strategy_name}.
2. Strategy Registry: A central store for functions (e.g., hash, mask) that can be extended.
3. Engine: A smart dispatcher that selects the best execution backend based on your input data.
4. Backends:
   • Pandas/Polars: Optimized for in-memory, vectorized operations.
   • Stream: Optimized for row-by-row processing (e.g., DB cursors) with constant memory usage.

Usage

The primary entry point is the anonymize() function, which automatically detects your data type and dispatches to the correct backend.

1. Unified Interface

import anonimize as nz
import pandas as pd
import polars as pl

schema = {
    "user_id": "hash",
    "email": "mask",
    "name": "faker_name"
}

# --- Auto-dispatch for Pandas ---
df_pd = pd.DataFrame(...)
result_pd = nz.anonymize(df_pd, schema)

# --- Auto-dispatch for Polars ---
df_pl = pl.DataFrame(...)
result_pl = nz.anonymize(df_pl, schema)

# --- Auto-dispatch for List of Dicts (Streams) ---
data = [{"user_id": 1, "email": "[e-mail]"}, ...]
result_iter_dicts = nz.anonymize(data, schema)

2. Database Cursors (Tuples)

For database drivers that yield tuples (like psycopg2 or sqlite3), use the columns argument or the specific from_cursor function.

cursor.execute("SELECT id, name, email FROM users")
columns = ["id", "name", "email"]

# Returns a generator of dicts
secure_rows = nz.anonymize(cursor, schema, columns=columns)

for row in secure_rows:
    print(row) # {'id': '...', 'name': '...', 'email': '...'}

3. Explicit Backend Selection (Advanced)

You can still strictly enforce a specific backend if desired:

from anonimize import from_pandas, from_polars, from_stream

# Will raise TypeError if input is not a Pandas DataFrame
nz.from_pandas(df, schema) 

Architecture

graph TD
    UserInput[Input Data + Schema] --> Main[anonymize()]
    Main --> Dispatcher{Check Type}
    
    Dispatcher -->|Polars DataFrame| PolarsExec[Polars Backend]
    Dispatcher -->|Pandas DataFrame| PandasExec[Pandas Backend]
    Dispatcher -->|Iterator/List| StreamExec[Stream Backend]
    
    PandasExec & PolarsExec & StreamExec --> Registry[StrategyRegistry]
    Registry --> Strategies[Anonymization Functions]
    Strategies -->|Hash/Mask/Fake| Result[Anonymized Data]

Extensibility

Register custom strategies globally:

from anonimize import register_strategy

def my_custom_redact(value):
    return "REDACTED" if value else None

register_strategy("redact", my_custom_redact)

# Now use it in your schema
schema = {"notes": "redact"}

Dependencies

• Core: faker, numpy
• Optional: pandas, polars

If an optional dependency is missing (e.g., passing a Pandas DataFrame without pandas installed), anonymize() will raise a clear TypeError.