Aller au contenu

Référence des utilitaires

Apprise inclut un ensemble de fonctions utilitaires réutilisées dans les plugins et les intégrations prises en charge. Les auteurs de plugins devraient privilégier ces helpers aux traitements ad hoc, car ils offrent un comportement cohérent, une meilleure gestion des cas limites et une journalisation plus sûre.

Cette page se concentre sur les helpers que vous utiliserez le plus souvent lorsque vous écrivez ou maintenez des plugins. D’autres utilitaires existent, mais ils ne sont volontairement pas couverts ici.

Les helpers sont exposés via apprise.utils (recommandé pour les auteurs de plugins) et implémentés dans les modules apprise.utils.*.

# Analyse et validation
from apprise.utils.parse import (
parse_bool,
parse_call_sign,
parse_emails,
parse_list,
parse_phone_no,
parse_url,
parse_urls,
is_call_sign,
is_email,
is_hostname,
is_ipaddr,
is_phone_no,
is_uuid,
validate_regex,
)
# Journalisation de debug plus sûre
from apprise.utils.sanitize import sanitize_payload
# Helpers d'encodage
from apprise.utils.base64 import (
base64_urlencode,
base64_urldecode,
encode_b64_dict,
decode_b64_dict,
)

Une bonne façon de voir les utilitaires d’analyse est la suivante :

  • les fonctions parse_*() découpent et normalisent des listes d’entrées, de façon souvent tolérante ;
  • beaucoup acceptent une option store_unparseable, ce qui vous permet de journaliser plus tard les valeurs rejetées ;
  • les fonctions is_*() valident une seule valeur candidate ;
  • elles renvoient False en cas d’échec, sinon des données normalisées que vous pouvez stocker en sécurité ;
  • validate_regex() est un helper strict servant à valider et, si besoin, reformater une valeur à l’aide d’une expression régulière.

La plupart des plugins utilisent les deux styles ensemble. Par exemple, un plugin SMS peut appeler parse_phone_no() pour découper une liste de cibles fournie par l’utilisateur, puis appeler is_phone_no() sur chaque entrée pour la valider et la normaliser.

Utilisez cette fonction dans vos implémentations de parse_url() ou parse_native_url() pour convertir une URL fournie par l’utilisateur en une structure normalisée. Elle gère la détection du schéma, l’échappement, l’extraction de l’utilisateur et du mot de passe, la vérification de l’hôte, l’analyse du port et celle de la chaîne de requête.

Utilisation typique :

from apprise.utils.parse import parse_url
results = parse_url(url, verify_host=False)
if not results:
return None
schema = results["schema"]
host = results.get("host")
qsd = results.get("qsd", {}) # dictionnaire des paramètres de requête
  • verify_host Lorsque cette valeur est True, parse_url() valide l’hôte et renvoie None s’il est absent ou invalide. C’est un bon comportement par défaut pour des cibles réseau. Lorsque cette valeur est False, la validation de l’hôte est assouplie, ce qui est utile pour des plugins qui traitent l’hôte comme un identifiant ou n’en ont pas besoin.

  • default_schema Utilisé quand l’utilisateur omet la portion schema://.

  • simple Lorsque cette valeur est True, la structure renvoyée est plus réduite. La plupart des plugins devraient conserver la valeur par défaut simple=False.

  • plus_to_space Contrôle si + dans la chaîne de requête devient un espace. Apprise le laisse par défaut à False, car + est fréquent dans les jetons et mots de passe.

Le dictionnaire renvoyé varie selon l’entrée, mais comprend généralement :

  • schema, et éventuellement host et port ;
  • user et password lorsqu’ils sont fournis ;
  • fullpath, et parfois path et query ;
  • les données de chaîne de requête analysées via qsd.

Convertit les représentations booléennes courantes en bool Python. C’est la méthode recommandée pour analyser des drapeaux issus d’une chaîne de requête ou d’une valeur de configuration.

from apprise.utils.parse import parse_bool
include_image = parse_bool(qsd.get("image"), default=False)
batch = parse_bool(qsd.get("batch"), default=True)
  • Faux : "0", "no", "off", "false", "deny", "disable", "never"
  • Vrai : "1", "yes", "on", "true", "allow", "enable"

Si une chaîne ne peut pas être interprétée, default est renvoyé. Pour les valeurs non textuelles, bool(value) est utilisé.

Parfois, vous souhaitez distinguer :

  • non fourni du tout ;
  • activé explicitement ;
  • désactivé explicitement.

Un cas classique est une fonctionnalité activée automatiquement seulement lorsqu’un autre réglage est présent, sauf si l’utilisateur la désactive explicitement.

raw = qsd.get("discovery") # None si non fourni
if raw is None:
# Non défini : choisir un défaut basé sur d'autres réglages
discovery = True if (self.secure and self.host) else False
else:
# Défini explicitement : respecter l'intention de l'utilisateur
discovery = parse_bool(raw, default=False)

Découpe des chaînes et entrées de type liste en une seule liste. Elle accepte plusieurs sources et les fusionne. Par défaut, le résultat est trié et dédoublonné.

from apprise.utils.parse import parse_list
tags = parse_list(qsd.get("tag"), cast=str)
targets = parse_list(qsd.get("to"), cast=str, allow_whitespace=False)
  • cast convertit les valeurs avant l’analyse lorsque c’est possible ;
  • allow_whitespace contrôle si les espaces sont traités comme séparateurs ;
  • sort contrôle si le résultat est normalisé en liste triée unique (True) ou renvoyé dans l’ordre d’analyse (False).

Ces helpers sont couramment utilisés dans les intégrations de type email et partout où l’utilisateur peut fournir plusieurs destinataires.

  • parse_emails() extrait plusieurs adresses candidates depuis des chaînes et parcourt récursivement tuples, listes et ensembles ;
  • is_email() valide une seule adresse et renvoie un résultat structuré.
from apprise.utils.parse import parse_emails, is_email
recipients = []
for candidate in parse_emails(qsd.get("to")):
result = is_email(candidate)
if not result:
self.logger.warning("Adresse email invalide ignorée (%s).", candidate)
continue
recipients.append(result["full_email"])

Astuce : is_email() renvoie un dictionnaire, pas seulement un booléen. Lorsqu’il est présent, vous pouvez utiliser des champs comme name, domain et full_email pour construire une représentation canonique et conserver les correspondances de noms pour plus tard.

Extrait des URL depuis des chaînes ou des entrées de type liste et peut conserver les valeurs impossibles à analyser afin d’aider au signalement d’erreurs.

from apprise.utils.parse import parse_urls
endpoints = parse_urls(qsd.get("endpoint"))

Si aucun élément valide n’est détecté et que store_unparseable=True, l’entrée est quand même découpée selon des séparateurs courants puis renvoyée. Cela vous permet de journaliser précisément quelles valeurs ont été rejetées au lieu de tout jeter silencieusement.

Schéma pratique dans un plugin :

emails = parse_emails(qsd.get("to"), store_unparseable=True)
valid = []
invalid = []
for entry in emails:
if is_email(entry):
valid.append(entry)
else:
invalid.append(entry)
for entry in invalid:
self.logger.warning("Email invalide ignoré : %s", entry)

Utilisés par les plugins SMS, voix et télécom.

parse_phone_no() découpe une liste de cibles en entrées candidates, puis is_phone_no() valide une entrée et renvoie soit False, soit un dictionnaire.

from apprise.utils.parse import parse_phone_no, is_phone_no
valid = []
invalid = []
for candidate in parse_phone_no(targets):
result = is_phone_no(candidate)
if result:
valid.append(f"+{result['full']}")
else:
invalid.append(candidate)
for candidate in invalid:
self.logger.warning("Numéro de téléphone invalide ignoré (%s).", candidate)

Le dictionnaire renvoyé par is_phone_no() inclut généralement :

  • full (chiffres seulement) ;
  • pretty (formatage lisible, lorsque disponible) ;
  • country, area, line (peuvent être vides).

Contraintes importantes :

  • le nombre de chiffres doit être compris entre min_len (10 par défaut) et 14 inclus ;
  • les séparateurs courants sont acceptés à l’entrée ; vous devriez stocker la forme normalisée.

Utilisés par APRS et les intégrations radioamateur.

parse_call_sign() découpe une liste d’indicatifs, puis is_call_sign() valide une seule valeur et renvoie False ou un dictionnaire.

from apprise.utils.parse import parse_call_sign, is_call_sign
targets = []
for candidate in parse_call_sign(qsd.get("to")):
result = is_call_sign(candidate)
if not result:
self.logger.warning("Indicatif invalide ignoré (%s).", candidate)
continue
targets.append(result["callsign"].upper())

is_call_sign() valide un seul indicatif et renvoie False ou un dictionnaire :

  • callsign (en majuscules) ;
  • ssid (chaîne pouvant être vide).

Un schéma pratique ressemble à celui des emails et téléphones :

  • analyser une liste avec parse_call_sign(...) ;
  • valider chaque entrée avec is_call_sign(...) ;
  • avertir pour les valeurs invalides sans faire échouer toute la configuration, sauf si le service exige au moins une cible valide.

Ces helpers sont souvent utilisés par les plugins pour valider tôt les entrées et produire des messages d’erreur clairs.

  • is_hostname(hostname, ipv4=True, ipv6=True, underscore=True) Valide les noms d’hôte et éventuellement les adresses IP. Il prend en charge les underscores lorsque underscore=True afin de couvrir des conventions pragmatiques de Docker et d’environnements locaux.

  • is_ipaddr(addr, ipv4=True, ipv6=True) Valide IPv4 et IPv6. Pour IPv6, la fonction renvoie l’adresse entourée de crochets ([addr]) afin de respecter les attentes de format d’URL.

  • is_uuid(value) Valide des chaînes UUID.

Ces helpers renvoient False lorsqu’une valeur est invalide. Sinon, ils renvoient une chaîne normalisée.

Valide une valeur par rapport à une expression régulière. En cas de succès, elle renvoie la valeur nettoyée ; sinon elle renvoie None.

from apprise.utils.parse import validate_regex
token = validate_regex(qsd.get("token"), r"^[A-Z0-9]{32}$", flags="i")
if not token:
self.logger.warning("Un jeton invalide a été fourni")
return None

Si votre plugin définit une regex de jeton de template, comme le font la plupart des plugins, vous pouvez la réutiliser directement pour valider les entrées, ce qui garde la définition à un seul endroit.

Exemple :

self.token = validate_regex(
token, *self.template_tokens["token"]["regex"]
)

Cela suppose que template_tokens["token"]["regex"] soit un tuple de la forme (regex, flags).

  • les regex sont mises en cache en interne après compilation, donc les validations répétées coûtent peu ;
  • flags peut être passé comme entier ou comme chaîne de caractères, par exemple "imx" ;
  • lorsque fmt est fourni, des groupes capturés nommés peuvent être réassemblés dans une chaîne normalisée.

Exemple avec formatage :

value = validate_regex(
value="prefix-123",
regex=r"^(?P<prefix>[a-z]+)-(?P<id>[0-9]+)$",
flags="i",
fmt="{prefix}:{id}",
)
# value vaut maintenant "prefix:123" (ou None si pas de correspondance)

Utilisez ce helper avant de journaliser des payloads de requête ou des détails de réponse, en particulier lorsque la structure peut inclure des pièces jointes, de gros blobs encodés ou une imbrication profonde. Il réduit la pollution des logs et évite de divulguer des valeurs sensibles ou très volumineuses, tout en conservant assez de structure pour dépanner.

from apprise.utils.sanitize import sanitize_payload
self.logger.debug("payload=%s", sanitize_payload(payload))

Vous n’avez pas besoin d’assainir chaque log de debug. sanitize_payload() est surtout utile lorsque :

  • un payload peut contenir des pièces jointes (base64, bytes, métadonnées de fichier) ;
  • une réponse peut contenir des objets volumineux inattendus ;
  • vous voulez des aperçus sûrs sans recopier de grosses valeurs dans les logs.

Si votre log de debug peut être coûteux, protégez l’appel pour qu’il ne s’exécute que lorsque le niveau DEBUG est activé.

import logging
from apprise.utils.sanitize import sanitize_payload
if self.logger.isEnabledFor(logging.DEBUG):
self.logger.debug("Payload: %s", sanitize_payload(payload))
  • laisse inchangées les primitives (None, booléens, nombres, etc.) ;
  • résume les longues chaînes avec un marqueur compact et des aperçus début/fin ;
  • résume les bytes avec un marqueur borné basé sur sha256 ;
  • parcourt en sécurité les dictionnaires et séquences imbriqués ;
  • détecte la récursion et évite une traversée infinie ;
  • applique des limites de profondeur et de nombre d’éléments pour éviter des logs gigantesques.

Schéma pratique :

  1. Construisez normalement votre payload de requête.
  2. Si le debug est activé, journalisez sanitize_payload(payload).
  3. Envoyez le payload original au service amont.

Certains environnements bénéficient de limites plus strictes, par exemple pour empêcher de grosses pièces jointes ou des blobs de type Base64 d’entrer dans les logs.

from apprise.utils.sanitize import SanitizeOptions, sanitize_payload
opts = SanitizeOptions(
max_str_len=64,
preview=8,
max_depth=4,
max_items=250,
)
self.logger.debug("payload=%s", sanitize_payload(payload, options=opts))

Lors d’un dépannage, évitez d’augmenter globalement ces limites. Il est plus sûr de les ajuster localement et temporairement autour d’une instruction de debug précise.

Helpers Base64 URL-safe pour travailler avec des octets.

from apprise.utils.base64 import base64_urlencode, base64_urldecode
encoded = base64_urlencode(b"abc") # "YWJj"
decoded = base64_urldecode(encoded) # b"abc"

Ces helpers sont stricts sur les types d’entrée. Ils renvoient None pour les valeurs non prises en charge.

Helpers de dictionnaire utiles lorsqu’une API attend des composants de payload encapsulés en Base64, souvent pour transporter en sécurité des données binaires. Les valeurs sérialisables en JSON sont converties en chaînes puis préfixées par b64:.

from apprise.utils.base64 import encode_b64_dict, decode_b64_dict
original = {"int": 1, "float": 2.3}
encoded, needs_decoding = encode_b64_dict(original)
# encoded == {"int": "b64:MQ==", "float": "b64:Mi4z"}
# needs_decoding vaut True
decoded = decode_b64_dict(encoded)
# decoded == original

Si l’encodage JSON échoue pour une valeur, le helper se rabat sur une conversion en chaîne et indiquera qu’aucun décodage n’est requis.

Questions ou commentaires ?

Documentation

Vous avez repéré une faute de frappe ou une erreur ? Signalez-la ou proposez une correction .

Problèmes Techniques

Vous rencontrez un problème avec le code ? Ouvrez un ticket sur GitHub :

Conçu avec amour depuis le Canada