Run a Database Copy with the Python Client

This page shows a database run from the Python client, from the connections to the results. Read Database Runs and Database Anonymization first for the concepts.

Database runs need the Python client octopize.avatar 1.30.0 or later.

Set Up

import os

from avatars import Manager, shared_entity

manager = Manager(api_key=os.environ["AVATAR_API_KEY"])
runner = manager.create_runner("crm-test-estate", seed=20260908)

Give the runner a seed. A run can be several jobs, one per group of tables, and they must agree about every shared entity.

Copy One Database

Declare the Source and the Target

The source and the target are names of connections. You never write a connection string in a run.

SOURCE, TARGET = "sample-crm", "sample-crm-test"

runner.add_database(source=SOURCE, target=TARGET)

To copy only some tables, pass tables=["customer", "invoice"]. Without it, the run copies every table it finds.

The target must already have the schema, and hold no rows: the Python client neither creates tables nor empties a target. See Requirements for the Target. The Python client also has no option to read only a subset of the rows.

Get the Proposed Plan

Schema discovery is a job. It reads the source, proposes a plan, and stops. Nothing is copied and nothing is anonymized.

plan = runner.proposed_plan()

print(plan.summary())
# 1 anonymize, 3 pseudonymize; 6 column(s) awaiting review

print(plan.needs_review())
# {'audit_log': ['actor_email'], 'country': ['name'],
#  'customer': ['address', 'email', 'full_name', 'phone']}

Read the proposals before you decide:

for proposal in plan.proposals():
    print(proposal)

# customer.email: EMAIL (by rules, NOT reviewed)
# customer.full_name: FULL_NAME (by rules, NOT reviewed)
# customer.city: ADDRESS (by rules, NOT reviewed)

A proposal made by rules comes from the name of the variable. A proposal made by model comes with a confidence: read each one.

The plan also shows the groups of linked tables. Each group is one job, named after the table that the most links touch:

print(plan.clustered())
# {'group-audit_log': ['audit_log'], 'group-customer': ['country', 'customer', 'invoice']}

A plan that you build by hand, with SchemaPlan.for_tables(), records no group: all its tables are in group-1.

Review the Plan

Each method records a decision and returns the plan, so you can chain them:

plan = (
    plan.confirm_all()                     # agree with the classifier
    .reject("customer", "city")            # disagree: keep the real cities
    .purge("audit_log")                    # keep the table, remove the rows
    .continue_sequence("customer", "id")   # continue numbering after the copied rows
)

plan.is_reviewed()  # True: the job will be accepted

Method	What it does
confirm(table, *columns)	agrees with the classifier for these columns
confirm_all()	agrees with every proposal. Read proposals() first.
reject(table, *columns)	keeps these variables unchanged (frozen). Recorded as a decision.
replace(table, column, strategy)	pseudonymizes a variable with the strategy you choose
anonymize_columns(table, *columns)	synthesizes the values from the distribution of the variable
freeze_columns(table, *columns)	keeps the values unchanged
drop_columns(table, *columns)	removes the variables from the copy
copy(*tables)	copies the tables unchanged
pseudonymize(*tables)	keeps the rows, replaces the personal values
anonymize(*tables)	anonymizes the tables with the Avatar method
purge(*tables)	keeps the tables, with no rows
ignore(*tables)	leaves the tables out of the copy
continue_sequence(table, column)	continues a sequence after the copied rows
restart_sequence(table, column)	restarts a sequence from the beginning (for a purged table)
start_sequence_at(table, column, n)	starts a sequence at n

A misspelled name is refused immediately, with the names that exist:

plan.purge("audit_logs")
# UnknownTableError: this plan has no table 'audit_logs';
#                    it has: audit_log, country, customer, invoice

A key variable takes no variable action. Asking is refused with the reason:

plan.freeze_columns("customer", "country_code")
# KeyColumnError: customer.country_code is a key, so it cannot be frozen ...

freeze_columns() and reject() keep real values in the copy. Use them only for a variable that identifies no individual: see Privacy Risk of Frozen Variables.

Declare Shared Entities

If several tables describe the same individuals, declare a shared entity so that each individual gets one replacement everywhere:

runner.add_database(
    source=SOURCE,
    target=TARGET,
    plan=plan,
    shared_entities=[
        shared_entity(
            "rt_person",
            tables=["customer", "invoice"],
            key="customer_id",
            replace=["email"],
        )
    ],
)

The replacement strategy comes from the name of the attribute: email gets realistic fake emails. Pass strategy= to choose another one.

Set the Parameters and Run

Each table that the run anonymizes needs its anonymization parameters, as in any other run. The plan knows which tables those are:

for table in plan.participating():
    runner.set_parameters(table, k=5)

runner.run()

Read the Results

The anonymized data is in the target database. There is nothing to download.

print(runner.statuses())
# {'group-customer/standard': 'finished', 'group-customer/signal_metrics': 'finished',
#  'group-customer/privacy_metrics': 'finished', 'group-customer/report': 'finished',
#  'group-audit_log/standard': 'finished', ...}

print(runner.cluster_sets())
# {'group-customer': '6acc42e5-...', 'group-audit_log': '28ef22bc-...'}

Each group is its own set of jobs, with its own metrics and reports. When the run has several groups, runner.get_all_results() does not wait for them. Read the results of each group from its own set:

for group, set_name in runner.cluster_sets().items():
    results = manager.create_runner_from_id(set_name).get_all_results()

A run with a single group is not split: runner.cluster_sets() is empty, and runner.get_all_results() returns its results. When the run declares shared entities and has several groups, statuses() also lists reference-tables/reference, the job that resolves them. Each job reports what it wrote: the tables, the number of rows, a foreign key that a purge left with no parent, and a table that the plan did not mention.

Copy Without a Plan

add_database() without plan= is a plain copy: every table, unchanged. Always pass a reviewed plan.

Copy Several Databases Together

Use this when the same individuals are in more than one database, for example a CRM and a billing system. If everything that is shared is in one database, use add_database().

Name each copy, so that a shared entity can say which database a table is in:

runner = manager.create_runner("full-test-estate", seed=20260908)

runner.add_databases(
    copies={
        "crm": ("crm_prod", "crm_test"),
        "billing": ("billing_prod", "billing_test"),
    }
)

plans = {
    "crm": runner.proposed_plan("crm").confirm_all().purge("audit_log"),
    "billing": runner.proposed_plan("billing").confirm_all(),
}

Then declare the shared entity with across. The identifying variable often has a different name in each database, so you name it per database:

runner.add_databases(
    copies={
        "crm": ("crm_prod", "crm_test"),
        "billing": ("billing_prod", "billing_test"),
    },
    plans=plans,
    shared_entities=[
        shared_entity(
            "rt_person",
            across={
                "crm": ("customer", "person_id"),
                "billing": ("invoice", "customer_ref"),
            },
            replace=["email"],
        )
    ],
)

runner.run()
runner.statuses()

In a run of several databases, each group is named after its copy, for example crm_group-customer and billing_group-invoice.

Before you run, compare the key values on both sides. If one side holds UUIDs and the other holds integers, you chose the wrong variable.

Every plan is checked and every target validated before a row is written. Each database is written by its own jobs, so one database can be written while another fails. A retry of the failed one is safe: it collects the same entities, in the same order, with the same seed, and gives the same replacements.

Manage Connections from Python

The connections that you see in the web client are also available from the Python client:

from avatars.models import ConnectionDraft, ConnectionPatch

connections = manager.auth_client.connections

# Which engines can this deployment reach?
for engine in connections.list_engines().engines:
    print(engine.label, engine.available, engine.preview)

# Which connections can I use?
for connection in connections.list_connections().connections:
    print(connection.name, connection.engine, connection.scope.value)

# Add a personal connection. Without scope="user", the API uses the
# organization scope, which only an administrator can create.
connections.put_connection(
    ConnectionDraft(
        engine="postgresql",
        label="Scratch database",
        scope="user",
        fields={
            "host": "db.internal",
            "port": "5432",
            "database": "scratch",
            "username": "avatar",
            "password": os.environ["SCRATCH_DB_PASSWORD"],
        },
    ),
    "scratch",
)

# Test it: reachable, has a schema, is empty
print(connections.test_connection("scratch"))
# reachable=True engine='postgresql' has_schema=True is_empty=False table_count=12

# Rename or relabel it, then delete it
connections.patch_connection(ConnectionPatch(label="Scratch (practice)"), "scratch")
connections.delete_connection("scratch")

list_engines() also describes the form fields of each engine (engine.fields). See Database Connections for the rules on scopes and credentials.

The client has four more calls on connections. The request models are in avatars.models.

Call	What it does
test_draft_connection(ConnectionDraft(...))	tests a connection before you store it
list_connection_presets()	lists the demo databases and the other ready-made connections that your installation offers
add_connection_presets(ConnectionPresetRequest(kind="demo", scope="user"))	adds the demo connections, as Add demo connections does
sample_connection(ConnectionSampleRequest(columns={...}, limit=5), name)	reads a few real values of some variables, to check that a key holds the same thing everywhere

sample_connection() reads 1 to 10 tables, 1 to 5 variables per table, and at most 20 values.

Tutorials

The Python client repository has three notebooks on database runs:

• Tutorial 10: a test database
• Tutorial 11: several databases at once
• Tutorial 12: database connections