Lancer une copie de base de données avec le client Python

Cette page montre une exécution sur base de données depuis le client Python, des connexions jusqu'aux résultats. Lisez d'abord Exécutions sur base de données et Anonymisation de bases de données pour les concepts.

Les exécutions sur base de données nécessitent le client Python octopize.avatar en version 1.30.0 ou ultérieure.

Mise en place

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)

Donnez une seed au runner. Une exécution peut comporter plusieurs jobs, un par groupe de tables, et ils doivent être d'accord sur chaque entité partagée.

Copier une base de données

Déclarer la source et la cible

La source et la cible sont des noms de connexions. Vous n'écrivez jamais de chaîne de connexion dans une exécution.

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

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

Pour ne copier que certaines tables, passez tables=["customer", "invoice"]. Sans ce paramètre, l'exécution copie toutes les tables qu'elle trouve.

La cible doit déjà posséder le schéma, et ne contenir aucune ligne : le client Python ne crée pas de tables et ne vide pas une cible. Voir Exigences pour la cible. Le client Python n'a pas non plus d'option pour ne lire qu'un sous-ensemble des lignes.

Obtenir le plan proposé

La découverte du schéma est un job. Elle lit la source, propose un plan, puis s'arrête. Rien n'est copié et rien n'est anonymisé.

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']}

Lisez les propositions avant de décider :

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)

Une proposition faite by rules (par règles) vient du nom de la variable. Une proposition faite by model (par modèle) est accompagnée d'un niveau de confiance : lisez chacune d'elles.

Le plan montre aussi les groupes de tables liées. Chaque groupe est un job, nommé d'après la table que touchent le plus de liens :

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

Un plan que vous construisez à la main, avec SchemaPlan.for_tables(), n'enregistre aucun groupe : toutes ses tables sont dans group-1.

Relire le plan

Chaque méthode enregistre une décision et renvoie le plan : vous pouvez donc les enchaîner.

plan = (
    plan.confirm_all()                     # accepter l'avis du classifieur
    .reject("customer", "city")            # refuser : garder les vraies villes
    .purge("audit_log")                    # garder la table, supprimer les lignes
    .continue_sequence("customer", "id")   # continuer la numérotation après les lignes copiées
)

plan.is_reviewed()  # True : le job sera accepté

Méthode	Ce qu'elle fait
confirm(table, *columns)	accepte l'avis du classifieur pour ces variables
confirm_all()	accepte toutes les propositions. Lisez d'abord proposals().
reject(table, *columns)	garde ces variables inchangées (gelées). Enregistré comme une décision.
replace(table, column, strategy)	pseudonymise une variable avec la stratégie de votre choix
anonymize_columns(table, *columns)	synthétise les valeurs à partir de la distribution de la variable
freeze_columns(table, *columns)	garde les valeurs inchangées
drop_columns(table, *columns)	retire les variables de la copie
copy(*tables)	copie les tables sans modification
pseudonymize(*tables)	garde les lignes, remplace les valeurs personnelles
anonymize(*tables)	anonymise les tables avec la méthode Avatar
purge(*tables)	garde les tables, sans aucune ligne
ignore(*tables)	exclut les tables de la copie
continue_sequence(table, column)	continue une séquence après les lignes copiées
restart_sequence(table, column)	redémarre une séquence depuis le début (pour une table purgée)
start_sequence_at(table, column, n)	fait commencer une séquence à n

Un nom mal orthographié est refusé immédiatement, avec la liste des noms qui existent :

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

Une variable de clé ne reçoit aucune action de variable. La demande est refusée, avec la raison :

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

freeze_columns() et reject() conservent les vraies valeurs dans la copie. Ne les utilisez que pour une variable qui n'identifie aucun individu : voir Risque pour la vie privée des variables gelées.

Déclarer des entités partagées

Si plusieurs tables décrivent les mêmes individus, déclarez une entité partagée pour que chaque individu reçoive un seul remplacement partout :

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

La stratégie de remplacement dépend du nom de l'attribut : email reçoit de faux e-mails réalistes. Passez strategy= pour en choisir une autre.

Régler les paramètres et lancer

Chaque table que l'exécution anonymise a besoin de ses paramètres d'anonymisation, comme dans toute autre exécution. Le plan sait de quelles tables il s'agit :

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

runner.run()

Lire les résultats

Les données anonymisées se trouvent dans la base de données cible. Il n'y a rien à télécharger.

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-...'}

Chaque groupe est son propre ensemble de jobs, avec ses propres métriques et rapports. Quand l'exécution a plusieurs groupes, runner.get_all_results() ne les attend pas. Lisez les résultats de chaque groupe depuis son propre ensemble :

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

Une exécution à un seul groupe n'est pas scindée : runner.cluster_sets() est vide, et runner.get_all_results() renvoie ses résultats. Quand l'exécution déclare des entités partagées et a plusieurs groupes, statuses() liste aussi reference-tables/reference, le job qui les résout. Chaque job indique ce qu'il a écrit : les tables, le nombre de lignes, une clé étrangère qu'une purge a laissée sans parent, et une table que le plan ne mentionnait pas.

Copie sans plan

add_database() sans plan= est une simple copie : toutes les tables, inchangées. Passez toujours un plan relu.

Copier plusieurs bases de données ensemble

Utilisez cette méthode quand les mêmes individus se trouvent dans plusieurs bases de données, par exemple un CRM et un système de facturation. Si tout ce qui est partagé se trouve dans une seule base, utilisez add_database().

Nommez chaque copie, pour qu'une entité partagée puisse indiquer dans quelle base se trouve une table :

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(),
}

Déclarez ensuite l'entité partagée avec across. La variable d'identification porte souvent un nom différent dans chaque base : vous la nommez donc pour chaque base.

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()

Dans une exécution sur plusieurs bases de données, chaque groupe est nommé d'après sa copie, par exemple crm_group-customer et billing_group-invoice.

Avant de lancer, comparez les valeurs de clé des deux côtés. Si un côté contient des UUID et l'autre des entiers, vous avez choisi la mauvaise variable.

Chaque plan est vérifié et chaque cible validée avant l'écriture de la moindre ligne. Chaque base de données est écrite par ses propres jobs : une base peut donc être écrite alors qu'une autre échoue. Relancer celle qui a échoué est sans risque : elle collecte les mêmes entités, dans le même ordre, avec la même graine, et produit les mêmes remplacements.

Gérer les connexions depuis Python

Les connexions que vous voyez dans le client web sont aussi disponibles depuis le client Python :

from avatars.models import ConnectionDraft, ConnectionPatch

connections = manager.auth_client.connections

# Quels moteurs ce déploiement peut-il atteindre ?
for engine in connections.list_engines().engines:
    print(engine.label, engine.available, engine.preview)

# Quelles connexions puis-je utiliser ?
for connection in connections.list_connections().connections:
    print(connection.name, connection.engine, connection.scope.value)

# Ajouter une connexion personnelle. Sans scope="user", l'API utilise la
# portée de l'organisation, que seul un administrateur peut créer.
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",
)

# La tester : joignable, avec un schéma, vide
print(connections.test_connection("scratch"))
# reachable=True engine='postgresql' has_schema=True is_empty=False table_count=12

# La renommer ou changer son nom affiché, puis la supprimer
connections.patch_connection(ConnectionPatch(label="Scratch (practice)"), "scratch")
connections.delete_connection("scratch")

list_engines() décrit aussi les champs de formulaire de chaque moteur (engine.fields). Voir Connexions aux bases de données pour les règles sur les portées et les identifiants.

Le client propose quatre autres appels sur les connexions. Les modèles de requête se trouvent dans avatars.models.

Appel	Ce qu'il fait
test_draft_connection(ConnectionDraft(...))	teste une connexion avant que vous l'enregistriez
list_connection_presets()	liste les bases de démonstration et les autres connexions prêtes à l'emploi que propose votre installation
add_connection_presets(ConnectionPresetRequest(kind="demo", scope="user"))	ajoute les connexions de démonstration, comme le fait Ajouter les connexions de démonstration
sample_connection(ConnectionSampleRequest(columns={...}, limit=5), name)	lit quelques valeurs réelles de certaines variables, pour vérifier qu'une clé contient la même chose partout

sample_connection() lit de 1 à 10 tables, de 1 à 5 variables par table, et au plus 20 valeurs.

Tutoriels

Le dépôt du client Python contient trois notebooks sur les exécutions sur base de données :

• Tutoriel 10 : une base de test
• Tutoriel 11 : plusieurs bases à la fois
• Tutoriel 12 : les connexions aux bases de données