- À propos de la bibliothèque io
- Flux en mémoire : StringIO et BytesIO
- Les classes de base et la hiérarchie
- Gestion des encodages avec TextIOWrapper
- Lecture et écriture bufferisées avec BufferedReader et BufferedWriter
- Interface avec les fichiers système et open()
- Cas pratique : création d'un pipeline de traitement de données
1. À propos de la bibliothèque io
La bibliothèque io (Input/Output) de Python fournit des outils pour gérer les flux de données de manière efficace et cohérente. Elle fait partie de la bibliothèque standard et offre une interface unifiée pour manipuler différents types de flux : fichiers texte, fichiers binaires et flux en mémoire comme les chaînes de caractères ou les bytes. L'utilisation principale de ce module est d'encapsuler les opérations de lecture/écriture brutes dans des objets plus faciles à manipuler, avec prise en charge de l'encodage, de la mise en mémoire tampon et des itérateurs.
Les classes principales sont organisées dans une hiérarchie : IOBase (la classe mère abstraite), RawIOBase (pour les opérations d'E/S brutes), BufferedIOBase (pour les flux avec tampon) et TextIOBase (spécialisée pour le texte). Les implémentations concrètes les plus utilisées sont BytesIO, StringIO, et les wrappers autour des descripteurs de fichier système.
2. Flux en mémoire : StringIO et BytesIO
Les classes StringIO et BytesIO permettent de traiter respectivement des chaînes de caractères (str) et des données binaires (bytes) comme des fichiers ouverts en mémoire. Cela est extrêmement utile pour les tests unitaires, la manipulation de données sans créer de fichiers physiques, ou l'interface avec des bibliothèques qui attendent un objet de type "fichier".
2.1. Utilisation de StringIO
StringIO crée un objet fichier en mémoire qui stocke une chaîne de caractères. On peut y écrire avec la méthode write() et lire avec read(), readline(), ou en itérant dessus, exactement comme avec un fichier ouvert avec open().
Voici un exemple de création, d'écriture et de lecture avec un objet StringIO :
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
from io import StringIO # Création d'un flux en mémoire flux_memoire = StringIO() # Écriture de données dans le flux flux_memoire.write("Bonjour le monde.\n") flux_memoire.write("Ceci est la seconde ligne.\n") # Important : se repositionner au début pour lire flux_memoire.seek(0) # Lecture de tout le contenu contenu = flux_memoire.read() print(contenu) # Sortie attendue : # Bonjour le monde. # Ceci est la seconde ligne. |
2.2. Utilisation de BytesIO
BytesIO fonctionne sur le même principe que StringIO, mais pour des données binaires. C'est l'outil idéal pour manipuler des images, des fichiers compressés, ou tout autre format binaire en mémoire.
Exemple d'écriture et de lecture de données binaires :
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
from io import BytesIO # Création d'un flux binaire en mémoire flux_binaire = BytesIO() # Écriture de bytes dans le flux flux_binaire.write(b'\x89PNG\r\n\x1a\n') En-tête fictif de fichier PNG flux_binaire.write(b'Donnees binaires...') # Retour au début flux_binaire.seek(0) # Lecture des premiers bytes premiers_bytes = flux_binaire.read(8) print(f"En-tête lu : {premiers_bytes}") # Sortie attendue : En-tête lu : b'\x89PNG\r\n\x1a\n' |
3. Les classes de base et la hiérarchie
Comprendre la hiérarchie des classes du module io est essentiel pour choisir la bonne classe en fonction des besoins. IOBase est la classe racine qui définit l'interface commune (comme close(), seek(), tell()). RawIOBase donne accès à un flux d'octets non bufferisé. BufferedIOBase ajoute une couche de tampon pour améliorer les performances. TextIOBase est spécialisée pour le texte et gère l'encodage/décodage.
L'exemple suivant illustre comment vérifier le type d'un flux et utiliser certaines méthodes de base :
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
from io import StringIO, BytesIO, TextIOBase, BufferedIOBase # Créer différents flux flux_texte = StringIO("du texte") flux_bytes = BytesIO(b"des bytes") # Vérifier les types et la hiérarchie print(f"flux_texte est-il une instance de TextIOBase ? {isinstance(flux_texte, TextIOBase)}") print(f"flux_bytes est-il une instance de BufferedIOBase ? {isinstance(flux_bytes, BufferedIOBase)}") # Utiliser une méthode commune de IOBase : closed print(f"flux_texte est fermé ? {flux_texte.closed}") flux_texte.close() print(f"flux_texte est fermé après close() ? {flux_texte.closed}") # Sortie attendue : # flux_texte est-il une instance de TextIOBase ? True # flux_bytes est-il une instance de BufferedIOBase ? True # flux_texte est fermé ? False # flux_texte est fermé après close() ? True |
4. Gestion des encodages avec TextIOWrapper
Lorsque vous travaillez avec du texte et des flux binaires bruts (comme un socket ou un fichier ouvert en mode binaire), vous devez souvent gérer l'encodage. La classe TextIOWrapper permet d'envelopper un flux binaire (BufferedIOBase) pour fournir une interface texte, avec encodage et décodage automatique.
Ceci est plus flexible que d'ouvrir un fichier avec open(mode='r') car vous pouvez changer l'encodage à la volée ou envelopper n'importe quel flux binaire.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
from io import BytesIO, TextIOWrapper # Créer un flux binaire en mémoire contenant du texte encodé en UTF-8 flux_binaire_source = BytesIO("Café à Paris".encode('utf-8')) flux_binaire_source.seek(0) # Envelopper ce flux binaire avec un TextIOWrapper pour le lire comme du texte wrapper_texte = TextIOWrapper(flux_binaire_source, encoding='utf-8') # Lire le texte via le wrapper texte_lu = wrapper_texte.read() print(f"Texte lu : {texte_lu}") # Il est aussi possible d'écrire du texte et de récupérer les bytes encodés flux_binaire_cible = BytesIO() wrapper_ecriture = TextIOWrapper(flux_binaire_cible, encoding='iso-8859-1') wrapper_ecriture.write("Héllö") wrapper_ecriture.flush() # Important pour vider le buffer dans le flux sous-jacent # Récupérer les bytes encodés selon l'encodage choisi flux_binaire_cible.seek(0) bytes_produits = flux_binaire_cible.read() print(f"Bytes produits (ISO-8859-1) : {bytes_produits}") # Sortie attendue : # Texte lu : Café à Paris # Bytes produits (ISO-8859-1) : b'H\xe9ll\xf6' |
5. Lecture et écriture bufferisées avec BufferedReader et BufferedWriter
Pour optimiser les performances, notamment avec des fichiers ou des flux réseaux, Python utilise la mise en mémoire tampon (buffering). Les classes BufferedReader et BufferedWriter du module io permettent de contrôler ce comportement explicitement.
Un BufferedReader lit par blocs (par exemple 8192 octets) depuis le flux brut, même si vous ne demandez qu'un octet à la fois, réduisant ainsi le nombre d'appels système coûteux.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
from io import BytesIO, BufferedReader, BufferedWriter # Créer un flux binaire brut contenant beaucoup de données donnees = b"x" * 10000 # 10000 bytes flux_brut = BytesIO(donnees) # L'envelopper dans un BufferedReader avec une taille de buffer personnalisée flux_bufferise_en_lecture = BufferedReader(flux_brut, buffer_size=2048) # Lire seulement 100 bytes. En interne, le buffer va probablement en lire 2048. premier_morceau = flux_bufferise_en_lecture.read(100) print(f"Longueur du premier morceau lu : {len(premier_morceau)}") # La position dans le flux brut a avancé de la taille du buffer lu print(f"Position dans le flux brut après lecture : {flux_brut.tell()}") # Exemple avec BufferedWriter flux_cible_brut = BytesIO() flux_bufferise_en_ecriture = BufferedWriter(flux_cible_brut, buffer_size=1024) # Écrire des données. Elles sont accumulées dans le buffer jusqu'à ce qu'il soit plein ou qu'on flush(). flux_bufferise_en_ecriture.write(b"a" * 500) print(f"Taille du flux cible avant flush : {len(flux_cible_brut.getvalue())}") flux_bufferise_en_ecriture.write(b"b" * 600) # Le buffer dépasse 1024, une partie est écrite flux_bufferise_en_ecriture.flush() # Force l'écriture de tout ce qui reste dans le buffer print(f"Taille du flux cible après flush : {len(flux_cible_brut.getvalue())}") # Sortie attendue (approximative, dépend de l'implémentation) : # Longueur du premier morceau lu : 100 # Position dans le flux brut après lecture : 2048 # Taille du flux cible avant flush : 0 # Taille du flux cible après flush : 1100 |
6. Interface avec les fichiers système et open()
La fonction built-in open() de Python utilise en interne les classes du module io pour retourner un objet fichier. Comprendre le lien entre les deux permet de mieux contrôler le comportement des fichiers, comme le buffering ou l'encodage.
Lorsque vous appelez open('fichier.txt', 'r', encoding='utf-8'), Python crée en réalité une chaîne d'objets io : un FileIO (RawIOBase) enveloppé dans un BufferedReader, lui-même enveloppé dans un TextIOWrapper.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
import io # Ouvrir un fichier avec open() - méthode standard with open('exemple_io.txt', 'w', encoding='utf-8') as f: # Vérifier la classe de l'objet retourné print(f"Type de l'objet fichier : {type(f)}") # Écrire quelque chose f.write("Écriture via open().") # Ré-ouvrir le fichier en mode binaire pour voir les couches with open('exemple_io.txt', 'rb') as f_binaire: # Lire les bytes bruts contenu_brut = f_binaire.read() print(f"Contenu brut : {contenu_brut}") # Recréer un TextIOWrapper manuellement autour de ce fichier binaire f_binaire.seek(0) wrapper_manuel = io.TextIOWrapper(f_binaire, encoding='utf-8') texte = wrapper_manuel.read() print(f"Texte décodé manuellement : {texte}") # Nettoyage : supprimer le fichier créé (nécessite le module os) import os if os.path.exists('exemple_io.txt'): os.remove('exemple_io.txt') # Sortie attendue : # Type de l'objet fichier : <class '_io.TextIOWrapper'> # Contenu brut : b'\xc3\x89criture via open().' # Texte décodé manuellement : Écriture via open(). |
7. Cas pratique : création d'un pipeline de traitement de données
Pour conclure, voici un cas pratique qui combine plusieurs concepts du module io. L'objectif est de simuler un traitement de données où des données textuelles sont générées, transformées (encodées, compressées de manière simulée), puis lues finalement. Nous utiliserons StringIO, BytesIO et TextIOWrapper.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 |
from io import StringIO, BytesIO, TextIOWrapper # Étape 1 : Génération de données textuelles dans un StringIO print("=== Étape 1 : Génération ===") donnees_texte = StringIO() for i in range(5): donnees_texte.write(f"Ligne de données n°{i} : valeur = {i*10}\n") donnees_texte.seek(0) print("Données générées :") print(donnees_texte.read()) # Étape 2 : Transformation (simulation d'encodage et de traitement binaire) print("\n=== Étape 2 : Transformation ===") donnees_texte.seek(0) # Lire le texte et l'encoder en bytes avec un encodage spécifique texte_complet = donnees_texte.read() bytes_encodes = texte_complet.encode('utf-16') # Placer ces bytes dans un BytesIO pour traitement flux_bytes_transforme = BytesIO(bytes_encodes) # Simuler un traitement : ajouter un marqueur de début marqueur = b'===DEBUT===' donnees_finales_bytes = BytesIO() donnees_finales_bytes.write(marqueur) donnees_finales_bytes.write(flux_bytes_transforme.read()) donnees_finales_bytes.seek(len(marqueur)) # Se placer après le marqueur # Étape 3 : Lecture finale print("\n=== Étape 3 : Lecture finale ===") # Envelopper la partie après le marqueur avec un TextIOWrapper pour décoder wrapper_final = TextIOWrapper(donnees_finales_bytes, encoding='utf-16') try: texte_final = wrapper_final.read() print("Données finales décodées :") print(texte_final) except UnicodeError as e: print(f"Erreur de décodage : {e}") # Sortie attendue (approximative, l'encodage utf-16 ajoute des BOM) : # === Étape 1 : Génération === # Données générées : # Ligne de données n°0 : valeur = 0 # Ligne de données n°1 : valeur = 10 # Ligne de données n°2 : valeur = 20 # Ligne de données n°3 : valeur = 30 # Ligne de données n°4 : valeur = 40 # # === Étape 2 : Transformation === # # === Étape 3 : Lecture finale === # Données finales décodées : # Ligne de données n°0 : valeur = 0 # Ligne de données n°1 : valeur = 10 # Ligne de données n°2 : valeur = 20 # Ligne de données n°3 : valeur = 30 # Ligne de données n°4 : valeur = 40 |
Auteur : Younes Derfoufi
Lieu de travail : CRMEF OUJDA
Site Web : www.tresfacile.net
Chaine YouTube : https://www.youtube.com/user/InformatiquesFacile
Me contacter : https://www.tresfacile.net/me-contacter/
