Création d'un fichier logique Python pour les applications de traitement

Pour créer une application de processus personnalisée, vous devez créer et télécharger un fichier logique Python. Le fichier logique Python extrait et transforme vos données, qui deviennent la source de données de votre projet d'exploration de processus.

Effectuez les étapes suivantes pour créer un fichier logique python :

  1. Téléchargez et ouvrez le modèle de fichier logique python
  2. Spécifier le point d'entrée.
  3. Spécifier les variables configurables.
  4. Facultatif : Définissez un calendrier pour les options de rafraîchissement des données.
  5. Spécifier le format des données de sortie.
  6. Facultatif : gérer les exceptions dans le code Python.

Spécification du point d'entrée

Le point d'entrée ou la fonction d'entrée sert d'entrée au processus de transformation des données. Sans point d'entrée défini dans le fichier logique, la transformation des données échoue avant même d'avoir commencé.

Il s'agit du point d'entrée dans le modèle de fichier logique python :

def execute(context):

Spécification des variables configurables

Des variables configurables calculent le processus de transformation des données. Il s'agit de variables dynamiques telles que token key, server ou systemURL, username, et password, et d'autres champs de saisie dynamiques.

Si vous spécifiez les variables, l'utilisateur pourra par la suite configurer les données du journal des événements lorsqu'il utilisera votre application de traitement. Pour plus d'informations, voir Définition des entrées utilisateur.

À partir de Process Mining 1.14.3, l'obtention de variables configurables en tant que variables d'environnement n'est plus prise en charge.

Vous pouvez spécifier des variables configurables dans une application de processus personnalisée en utilisant un objet configurable parmi les objets contextuels. Les variables de configuration sont disponibles dans un dictionnaire Python que vous pouvez récupérer en utilisant la syntaxe context.config . Dans l'exemple suivant, vous définissez la fonction execute en tant que paramètre context :

def execute(context): 
config = context[“config”] 
# example retrieving an input property named ‘URL’ 
url = config[“URL”]

Si la source de données brutes que l'utilisateur de votre application de processus télécharge est au format .CSV , vous pouvez configurer le fichier logique Python pour transformer les données en source pour votre projet d'exploration de processus. Avant de télécharger les données, l'utilisateur doit transformer le fichier du format .CSV au format .zip . Ensuite, le fichier logique Python accède au fichier .zip pour effectuer la transformation nécessaire des données dans le même répertoire.

Le fichier logique peut également récupérer le nom du fichier téléchargé .zip à partir du contexte JSON qui est transmis à la fonction execute du fichier logique. Le nom de la clé du fichier logique téléchargé est fileUploadName et la valeur est le nom du fichier téléchargé .zip . Par exemple, si le nom du fichier .zip téléchargé est "Input_sample.zip", le fichier logique peut récupérer le nom du fichier .zip en utilisant la commande suivante :

def execute(context): 
# Example retrieving the name of an uploaded zip file 
myFileUploadName = context[“fileUploadName”]

Optionnel : Définition d'un calendrier pour les options de rafraîchissement des données

En tant qu'utilisateur d'une application de processus personnalisée, vous disposez de deux options de planification pour gérer les mises à jour des données du journal des événements de votre processus :

  • La méthode de remplacement, qui consiste à remplacer la source de données existante. La source de données du projet précédent n'est pas prise en compte. La nouvelle source de données est traitée en fonction de vos besoins : vous pouvez choisir la fréquence de remplacement de vos données ou utiliser un calendrier pour fixer une date préférée pour le remplacement des données.

    Si vous souhaitez uniquement utiliser la méthode de remplacement, vous ne devez pas apporter de modifications supplémentaires à votre fichier logique Python.

  • La méthode de mise à jour incrémentale, où les nouvelles données sont ajoutées à la source de données existante du projet. Les données précédentes et nouvelles sont utilisées dans votre projet.

    Si vous souhaitez utiliser la méthode de mise à jour incrémentielle, vous devez configurer le fichier logique Python associé pour qu'il génère une planification permettant de récupérer de nouvelles données à partir d'une source externe. Pour ce faire, spécifiez deux variables clés, lastScheduleExecutionDate et lastScheduleExecutionTimezone, qui sont fournies dans le fichier logique Python.

    Les variables lastScheduleExecutionDate et lastScheduleExecutionTimezone sont essentielles pour les opérations logiques au sein du script Python, permettant la récupération incrémentale de nouvelles données du journal des événements en fonction de la date et du fuseau horaire de la dernière exécution réussie de la programmation.

    Vous pouvez accéder aux variables lastScheduleExecutionDate et lastScheduleExecutionTimezone par l'intermédiaire de l'objet contextuel. Ces variables sont accessibles dans le fichier logique Python à l'aide du dictionnaire contextuel. Elle est transmise à la logique Python par l'intermédiaire de la fonction execute :

    last_schedule_execution_date = context["lastSuccessfulExecution"]["lastScheduleExecutionDate"]
last_schedule_execution_timezone = context["lastSuccessfulExecution"]["lastScheduleExecutionTimezone"]
…
# Subsequent code using last_schedule_execution_date and last_schedule_execution_timezone
if last_schedule_execution_date and last_schedule_execution_timezone:
   # Logic using the last scheduled execution date and timezone

Analyse des données de last_schedule_execution au format ISO 8601

La date et l'heure de la dernière exécution programmée (c'est-à-dire la date de la dernière exécution programmée) sont au format ISO 8601 (aaaa-MM-jj 'T' HH:mm:ss.SSS ). Après avoir récupéré les données et le fuseau horaire comme indiqué dans l'exemple précédent, vous pouvez analyser les données de last_schedule_execution en chaîne de format ISO 8601 de la manière suivante :

from datetime import datetime
import zoneinfo

...

timezone = zoneinfo.ZoneInfo(last_schedule_execution_timezone )

# Parse the ISO 8601 formatted date string
date = datetime.fromisoformat(last_schedule_execution_date ).astimezone(timezone)
Note : Pour le traitement des fuseaux horaires, il est recommandé d'utiliser le module Python zoneinfo .

Si le fichier logique est destiné à faire de l'incrémentation, ces variables sont déclarées globalement de la manière suivante :


LAST_SCHEDULE_EXECUTION_DATE = None
LAST_SCHEDULE_EXECUTION_TIMEZONE = None

...

def execute(context):
   LAST_SCHEDULE_EXECUTION_DATE = context["lastSuccessfulExecution"]["lastScheduleExecutionDate"]
   LAST_SCHEDULE_EXECUTION_TIMEZONE = context["lastSuccessfulExecution"]["lastScheduleExecutionTimezone"]

Étant donné que le service Python appelle la fonction execute , il transmet le contexte à sa fonction d'exécution et, en conséquence, LAST_SCHEDULE_EXECUTION_DATE et LAST_SCHEDULE_EXECUTION_TIMEZONE sont activés.

Pour le premier ETL, les valeurs LAST_SCHEDULE_EXECUTION_DATE et LAST_SCHEDULE_EXECUTION_TIMEZONE doivent être None. Ainsi, quel que soit l'endroit où vous utilisez ces variables dans votre logique. Le premier ETL est exécuté sans les valeurs et un contrôle peut être utilisé comme suit :

if LAST_SCHEDULE_EXECUTION_DATE and LAST_SCHEDULE_EXECUTION_TIMEZONE:
   timezone = zoneinfo.ZoneInfo(last_schedule_execution_timezone )
   date = datetime.fromisoformat(last_schedule_execution_date ).astimezone(timezone)
Remarque : si vous envoyez ce format de date à une API distante, la date analysée précédente suppose que le format correspond aux attentes de l'API distante. Si ce n'est pas le cas, vous pouvez utiliser l'analyse pour convertir le format attendu par l'API distante.

Spécification du format des données de sortie

Le point d'entrée, que vous définissez dans le fichier logique, renvoie le résultat du processus de transformation des données. Le format de sortie doit être un objet Python DataFrame , qui est une structure de données permettant de manipuler et de stocker des données dans Python. La sortie du processus de transformation des données dans le fichier logique doit être soit pandas , soit polars, qui sont les deux modules DataFrame . Dans l'exemple suivant, vous configurez l'objet DataFrame :

   def output(event_list): 
      return pd.DataFrame(event_list) 
   def execute(context): 
      extract_and_transform(context) 
      return output(event_list)
Remarque : dans l'exemple, vous devez modifier les fonctions en fonction de vos besoins pour le fichier logique.

Gestion des exceptions dans le code Python

Pour informer l'utilisateur de l'application de processus de l'invalidité des identifiants ou de la perte de connexion à un système externe, vous pouvez configurer le code Python pour qu'il envoie une exception dédiée nommée ProcessAppException dans l'interface utilisateur de l'application de processus. L'extrait de code suivant donne un exemple de configuration de ProcessAppException :

   from process_app import ProcessAppException 
   def execute(context): 
      # Example raising a ProcessAppException 
      raise ProcessAppException("cannot connect to the system")