La puissance de Python intégré dans l'interopérabilité - un cas d'utilisation de MongoDB CDC

L'utilisation de Embedded Python lors de la création d'une solution basée sur InterSystems peut ajouter des fonctionnalités très puissantes et approfondies à votre boîte à outils.

J'aimerais partager un exemple de cas d'utilisation que j'ai rencontré : l'activation d'un CDC (Change Data Capture) pour une collection mongoDB - la capture de ces modifications, leur digestion via un flux d'interopérabilité et, pour finir, la mise à jour d'un EMR via une API REST.

La fonctionnalité principale – « Observation des changements » 👀

L'idée de base est d'utiliser le package PyMongo, qui, entre autres, permet de suivre les modifications au sein d'une base de données mongo via son change_streams.

La partie importante se trouve dans l'adaptateur entrant que j'ai créé et qui possède une méthode OnInit() qui inclut cette ligne :

self.changeStream = client.get_database(...).get_collection(...).watch()

J'utilise la méthode watch() ici pour surveiller les changements d'une collection spécifique dans une base de données.

Je les ai définis via les paramètres du service métier :

Ensuite, dans la méthode OnTask(), j'utilise ceci pour capturer les modifications :

while self.changeStream.alive:
    change = self.changeStream.try_next()
    if change is not None:

Ici, j'utilise la propriété alive et la méthode try_next() pour obtenir les détails des modifications survenues.

Pour chaque modification, je crée un objet IRIS Stream avec le contenu JSON et je l'envoie à la méthode ProcessInput() du service métier.

Dans le service métier, je crée une requête importée à partir du contenu JSON et je l'envoie au routeur.

Le flux général peut être illustré comme suit :

Voici une démonstration vidéo :

https://www.youtube.com/embed/Z_6rq-bhB_s
[Ceci est un lien intégré, mais vous ne pouvez pas consulter le contenu intégré directement sur le site car vous avez refusé les cookies nécessaires pour y accéder. Pour afficher le contenu intégré, vous devez accepter tous les cookies dans vos Paramètres des cookies]

Quelques détails techniques supplémentaires 👩‍💻

Le service métier doit « masser » un peu le JSON car il inclut du JSON « flou », avant que je puisse autoriser son importation en tant que JSON valide.

Par exemple, un message peut ressemble à ceci:

{
   "_id": {
      "_data": "82635019A0000000012B042C0100296E5A1004AB1154ACACD849A48C61756D70D3B21F463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F69640064635019A078BE67426D7CF4D2000004"
    },
    "operationType": "insert",
    "clusterTime": Timestamp({ "t": 1666193824, "i": 1 }),
    "collectionUUID": new UUID("ab1154ac-acd8-49a4-8c61-756d70d3b21f"),
    "wallTime": ISODate("2022-10-19T15:37:04.604Z"),
    "fullDocument": {
       "_id": ObjectId("635019a078be67426d7cf4d2"'),
       "name": "Giovanni Verga"
    },
    "ns": {
       "db": "test",
       "coll": "names"
    },
    "documentKey": {
       "_id": ObjectId("635019a078be67426d7cf4d2")
    }
}

Vous pouvez voir par exemple que Timestamp et new UUID ne sont pas des JSON valides.

Vous pouvez voir que le « Document » actuel se trouve dans la partie fullDocument, et pour cet exemple j'ai utilisé un schéma spécifique.

Mon document ressemble à ceci :

 

Exemple de document JSON pour cette démo

 

{
    "mrn":"123456789",
    "first_name":"john",
    "last_name":"smith",
    "dob":"2001-01-01",
    "gender":"male"
}

Dans votre cas, vous pouvez soit modifier cela pour l'adapter à votre schéma, soit même envisager d'adapter mon exemple pour utiliser une approche plus « dynamique » où vous pouvez définir un paramètre qui serait utilisé pour importer dynamiquement le JSON vers un nom de classe variable (par rapport à mon nom codé en dur).

[Dans tous les cas, pour charger votre schéma JSON et créer une classe à partir de celui-ci, vous pouvez utiliser la classe Sample.OpenAPIClassGenerator que j'ai incluse dans mon exemple (adapté du générateur de classe de définition OpenAPI de @Guillaume Rongier (et appeler la méthode ProcessFile() que j'ai ajoutée sur votre fichier de schéma JSON).]

Il s’agit essentiellement de la partie principale de cette fonctionnalité : Adaptateur + Service métier qui envoie des messages avec les modifications apportées à une collection de bases de données Mongo.

Élargir le tableau 👩‍⚕️

À des fins de démonstration et pour créer un flux complet qui « raconte une histoire » (ce qui était le cas d'utilisation réel que j'avais), j'ai également ajouté une cible pour ce CDC, qui est une API REST d'un Dossier Patient Informatisé (DPI ou EMR pour Electronic Medical Record) fictif, qui prend les données des patients et les insère dans une table de patients.

Vous trouverez cette partie sous le package Demo.EMR (Data.Patient pour la table et Util.API pour l'API REST) ; et le package Demo.Int pour la partie Opérations métiers.

En fait, vous pouvez réutiliser cette partie pour tout autre besoin que vous pourriez rencontrer lorsque vous avez besoin d'une API fictive pour envoyer des données de patient à des fins de test en tant que destination cible.

[Notez que la table « Mock EMR » et l'API incluent des éléments de données que je n'ai pas utilisés dans cette démonstration, comme l'adresse e-mail et le numéro de téléphone]

Un exemple de flux complet (avec captures d'écran) 📷

Voici donc un exemple de déroulement de la manière dont cela fonctionne :

1. Ajoutez un document à votre collection mongo.

 

Commandes Shell (Docker exec pour se connecter au conteneur mongo et commandes mongosh pour lancer ReplicaSet et un document vers une collection)

 

$ docker exec -it mongo1 bash
root@e3b93abb5596:/# mongosh
Current Mongosh Log ID: 676708e5955c5e4448fe6910
Connecting to:          mongodb://127.0.0.1:27017/?directConnection=true&serverSelectionTimeoutMS=2000&appName=mongosh+2.3.2
Using MongoDB:          6.0.19
Using Mongosh:          2.3.2

For mongosh info see: https://www.mongodb.com/docs/mongodb-shell/
test> use admin
switched to db admin
{ ok: 1 }
admin> db.auth("admin","admin123")
{ ok: 1 }
admin> rs.initiate()
{
  info2: 'no configuration specified. Using a default configuration for the set',
  me: '960427582b78:27017',
  ok: 1
}
rs0 [direct: other] admin> use testDB
switched to db testDB
rs0 [direct: primary] testDB> db.testCol.insertOne({"mrn":"123456789","first_name":"john","last_name":"smith","dob":"2001-01-01","gender":"male"})
{
  acknowledged: true,
  insertedId: ObjectId('676714119af8c381e2fe6914')
}

Lorsque vous insérez le document, grâce au « watcher » exécuté dans IRIS, celui-ci se déclenchera dans l'adaptateur et le service métier

2. Examinez votre table « EMR » et voyez les nouvelles données de Mongo insérées dans celle-ci

3. Examinez les « coulisses » d’InterSystems IRIS en prenant soin de ce flux

Faites l'essai de votre côté🏎

L'application Open Exchange inclut tout le code associé et les instructions de configuration pour une démonstration complète basée sur le conteneur Docker comme indiqué ci-dessus (avec toutes les parties associées, y compris les conteneurs Mongo).

[Remarque : Mongo a une notion de ReplicaSet et c'est seulement avec cette fonctionnalité que change_streams fonctionne. Pour plus de simplicité, la démonstration ci-dessus suppose que mongo1 (il y a aussi mongo2 et mongo3) est le « principal »].

Encore une fois, ce n’est qu’un exemple de la manière dont Embedded Python peut vous permettre de relever vos défis d’interopérabilité très rapidement et facilement.