Python – Bases pour Data Engineers

Ce notebook donne les bases de Python nécessaires pour la suite du parcours Data Engineering From Zero to Hero.

⚠️ Version Python recommandée

Version Statut Recommandation
Python 3.12 ✅ Dernière stable Recommandée pour nouveaux projets
Python 3.11 ✅ Stable Excellent choix, très performant
Python 3.10 ✅ Supportée Minimum pour les type hints modernes
Python 3.9 ⚠️ Maintenance Éviter pour nouveaux projets
Python 3.8 et avant ❌ Obsolète Ne pas utiliser

💡 Pour ce cours, utilise Python 3.11 ou 3.12.

Prérequis

Niveau Compétence
✅ Requis Avoir suivi le module 03_git_for_data_engineers
✅ Requis Savoir utiliser un terminal (Bash)
🟡 Optionnel Notions de programmation dans un autre langage

Objectifs du module

À la fin de ce notebook, tu seras capable de :

  • Installer et configurer ton environnement Python
  • Créer et gérer des environnements virtuels
  • Manipuler les types de base (nombres, chaînes, listes, dictionnaires)
  • Utiliser les conditions et boucles
  • Écrire des fonctions et des classes simples
  • Gérer les erreurs avec try / except
  • Lire et écrire des fichiers (texte, JSON)
  • Utiliser le module logging pour tracer un script

Pourquoi Python pour le Data Engineering ?

Raison Détail
Écosystème riche Pandas, PySpark, Airflow, dbt, FastAPI…
Polyvalent Scripts, APIs, pipelines, ML, automation
Standard de l’industrie Utilisé par Netflix, Spotify, Airbnb…
Facile à apprendre Syntaxe claire et lisible
Intégrations Connecteurs pour toutes les bases de données

0. Installation & Environnement 🛠️

Cette section explique comment installer Python, VS Code, Jupyter et vérifier que tout fonctionne. Les commandes sont données pour Windows, Linux et macOS.

0.1 Installer Python

Sous Windows

  1. Aller sur le site officiel : https://www.python.org/downloads/
  2. Télécharger la dernière version stable de Python 3.x.
  3. Lors de l’installation :
    • Cocher “Add Python to PATH” en bas de la première fenêtre ;
    • puis cliquer sur Install Now.

Sous Linux (Ubuntu / Debian)

sudo apt update
sudo apt install -y python3 python3-pip

Sous macOS

  • Option 1 : paquet officiel :
    • Télécharger un .pkg depuis https://www.python.org/downloads/mac-osx/
    • Installer comme une application classique.
  • Option 2 (si Homebrew est installé) :
brew install python

0.2 Vérifier l’installation de Python

Ouvrir un terminal (ou PowerShell sous Windows) puis taper :

python --version   # ou parfois: python3 --version

Tu dois voir une version du type : Python 3.12.x.

⚠️ Erreurs fréquentes : - python n’est pas reconnu → Python n’est pas dans le PATH ; - sur Linux/macOS, il faut parfois utiliser python3 au lieu de python.

0.3 Installer Visual Studio Code (VS Code)

  1. Télécharger VS Code : https://code.visualstudio.com/
  2. Installer la version adaptée à ton système (Windows, Linux, macOS).
  3. Lancer VS Code.

VS Code servira à : - éditer des scripts .py ; - ouvrir des notebooks Jupyter (.ipynb) ; - organiser un projet de data engineering complet.

0.4 Extensions VS Code : Python & Jupyter

Dans VS Code, aller dans l’onglet Extensions (icône de blocs à gauche), puis :

  1. Rechercher “Python” (éditeur : Microsoft) et l’installer ;
  2. Rechercher “Jupyter” (éditeur : Microsoft) et l’installer.

Ensuite, ouvrir un fichier .py ou .ipynb : VS Code proposera de sélectionner un interpréteur Python (en bas à droite). Choisir ton installation Python 3.x.

0.5 Jupyter Notebook — Installation et utilisation

Jupyter Notebook est un environnement interactif qui permet d’écrire du code, de l’exécuter, et de voir les résultats immédiatement. C’est l’outil idéal pour apprendre, explorer des données et documenter son travail.

💡 Ce cours est lui-même un Notebook Jupyter (fichier .ipynb) !

Installation

# Installer Jupyter
pip install notebook

# Ou avec Anaconda (déjà inclus)
# Rien à faire, c'est installé par défaut

Lancer Jupyter Notebook

# Dans ton terminal, place-toi dans ton dossier de travail
cd /chemin/vers/mon/projet

# Lancer Jupyter
jupyter notebook

Ce qui se passe :

  1. Un serveur local démarre
  2. Ton navigateur s’ouvre automatiquement sur http://localhost:8888
  3. Tu vois la liste des fichiers de ton dossier
┌─────────────────────────────────────────────────────────────────┐
│  Jupyter                                    [Quit] [Logout]     │
├─────────────────────────────────────────────────────────────────┤
│  Files    Running    Clusters                                   │
├─────────────────────────────────────────────────────────────────┤
│  data/                                                       │
│  01_intro.ipynb                                              │
│  02_basics.ipynb                                             │
│  script.py                                                   │
│                                                                 │
│  [New ▼]  ← Cliquer ici pour créer un nouveau notebook          │
└─────────────────────────────────────────────────────────────────┘

Créer un nouveau Notebook

  1. Cliquer sur New (en haut à droite)
  2. Sélectionner Python 3 (ou Python 3 (ipykernel))
  3. Un nouveau notebook s’ouvre : Untitled.ipynb
  4. Cliquer sur “Untitled” pour le renommer (ex: mon_premier_notebook.ipynb)

L’interface Jupyter

┌─────────────────────────────────────────────────────────────────┐
│  mon_notebook.ipynb                              [Trusted]      │
├─────────────────────────────────────────────────────────────────┤
│  File  Edit  View  Insert  Cell  Kernel  Help                   │
│  [💾] [+] [✂️] [📋] [▶️ Run] [⏹️] [🔄]    | Code ▼ |             │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  In [1]: █                              ← Cellule de code       │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘
Élément Description
Cellule Bloc où tu écris du code ou du texte
In [1]: Numéro d’exécution de la cellule
▶️ Run Exécute la cellule sélectionnée
Code ▼ Type de cellule (Code ou Markdown)

⌨Raccourcis clavier essentiels

Raccourci Action
Shift + Enter ▶️ Exécuter la cellule et passer à la suivante
Ctrl + Enter Exécuter la cellule (rester dessus)
Esc Passer en mode commande (cellule bleue)
Enter Passer en mode édition (cellule verte)
A (mode commande) Insérer une cellule au-dessus
B (mode commande) Insérer une cellule en-dessous
DD (mode commande) Supprimer la cellule
M (mode commande) Convertir en cellule Markdown
Y (mode commande) Convertir en cellule Code
Ctrl + S Sauvegarder le notebook

💡 Astuce : Apprends Shift + Enter en premier — c’est le raccourci que tu utiliseras le plus !

Premier test dans Jupyter

  1. Dans une cellule, tape :
print("Hello Data Engineer !")

  1. Appuie sur Shift + Enter

  2. Tu dois voir :

In [1]: print("Hello Data Engineer !")

Hello Data Engineer !

Types de cellules

Type Usage Exemple
Code Exécuter du Python print("Hello")
Markdown Documenter, titres, explications # Mon titre 
# Titre principal
## Sous-titre

Du texte en **gras** et en *italique*.

- Liste à puces
- Autre élément

🛑 Arrêter Jupyter

  1. Sauvegarder ton notebook (Ctrl + S)
  2. Fermer l’onglet du navigateur
  3. Dans le terminal où Jupyter tourne : appuyer sur Ctrl + C deux fois
^C
Shutdown this notebook server (y/[n])? y

💡 Alternative : Jupyter dans VS Code

Tu peux aussi ouvrir des notebooks directement dans VS Code (avec l’extension Jupyter installée) :

  1. Ouvrir VS Code
  2. File > Open File → sélectionner un fichier .ipynb
  3. Ou créer un nouveau fichier avec l’extension .ipynb

C’est souvent plus pratique car tu as tout dans le même éditeur !

0.6 Premier test Python

Python peut s’utiliser de 3 façons différentes :

Mode Usage Commande
Interactif Tester rapidement du code python ou python3
Script Exécuter un fichier .py python mon_script.py
Notebook Exploration, visualisation Jupyter / VS Code

Mode interactif (REPL)

REPL = Read-Eval-Print Loop (Lire-Évaluer-Afficher en boucle)

1. Lancer l’interpréteur Python :

# Dans ton terminal (PowerShell, CMD, Bash...)
python
# ou sur Linux/macOS
python3

2. Tu verras apparaître le prompt >>> :

Python 3.12.0 (main, Oct  2 2024, 12:00:00)
[GCC 11.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>>

3. Taper du code Python directement :

>>> print("Hello Data Engineer")
Hello Data Engineer

>>> 2 + 3
5

>>> nom = "Alice"
>>> print(f"Bonjour {nom}")
Bonjour Alice

4. Quitter l’interpréteur :

>>> exit()

Ou utiliser le raccourci clavier : - Windows : Ctrl + Z puis Entrée - Linux/macOS : Ctrl + D

Quand utiliser le mode interactif ?

✅ Bon usage ❌ Mauvais usage
Tester une syntaxe rapidement Écrire un programme complet
Vérifier le résultat d’une expression Code qu’on veut sauvegarder
Explorer une librairie (help(fonction)) Pipeline de production
Calculatrice avancée Travail collaboratif

💡 Pour ce cours, tu utiliseras surtout les Notebooks Jupyter (exploration) et les scripts .py (production). Le mode interactif est utile pour des tests rapides.

0.7 Environnements virtuels

Un environnement virtuel isole les dépendances de chaque projet. C’est indispensable en Data Engineering pour éviter les conflits de versions.

Pourquoi utiliser un environnement virtuel ?

❌ Sans environnement virtuel ✅ Avec environnement virtuel
Tous les projets partagent les mêmes packages Chaque projet a ses propres packages
Conflits de versions Isolation complète
Difficile à reproduire Reproductible avec requirements.txt

Option 1 : venv (intégré à Python)

# Créer un environnement virtuel
python -m venv mon_env

# Activer l'environnement
# Windows
mon_env\Scripts\activate

# Linux / macOS
source mon_env/bin/activate

# Tu verras (mon_env) au début de ta ligne de commande

# Désactiver l'environnement
deactivate

Option 2 : conda (Anaconda/Miniconda)

# Créer un environnement
conda create -n mon_projet python=3.11

# Activer
conda activate mon_projet

# Désactiver
conda deactivate

# Lister les environnements
conda env list

💡 Recommandation pour Data Engineers

Outil Quand l’utiliser
venv Projets Python purs, légers, CI/CD
conda Data Science, dépendances complexes (NumPy, Spark)

0.8 Gestion des packages avec pip

pip est le gestionnaire de packages Python. Tu l’utiliseras pour installer les librairies Data Engineering.

Commandes essentielles

# Installer un package
pip install pandas

# Installer une version spécifique
pip install pandas==2.0.0

# Installer plusieurs packages
pip install pandas numpy requests

# Mettre à jour un package
pip install --upgrade pandas

# Désinstaller
pip uninstall pandas

# Lister les packages installés
pip list

# Voir les infos d'un package
pip show pandas

Le fichier requirements.txt

Ce fichier liste toutes les dépendances d’un projet. Indispensable pour la reproductibilité.

# Générer le fichier à partir de l'environnement actuel
pip freeze > requirements.txt

# Installer toutes les dépendances d'un projet
pip install -r requirements.txt

Exemple de requirements.txt pour Data Engineering

# Data Processing
pandas>=2.0.0
numpy>=1.24.0
pyarrow>=12.0.0

# APIs
requests>=2.28.0
fastapi>=0.100.0

# Database
sqlalchemy>=2.0.0
psycopg2-binary>=2.9.0

# Testing
pytest>=7.0.0

💡 Bonne pratique : Toujours travailler dans un environnement virtuel avant d’installer des packages !

1. Variables et Types

Une variable est un nom qui référence une valeur en mémoire.

Python possède plusieurs types intégrés (builtins). Voici ceux à maîtriser absolument en Data Engineering :

Catégorie Type Exemple Usage Data Engineering
Numérique int 3, 42, -5 Comptage, index, tailles, IDs
float 3.14, 0.99 Prix, mesures, statistiques
Texte str "Abidjan" Parsing CSV/JSON, nettoyage, logs
Booléen bool True, False Conditions, filtres, validation
Séquences ordonnées list [1,2,3] Lignes CSV, séries numériques
tuple (200, "OK") Valeurs fixes, clés composites
Mapping dict {"id":1,"ville":"Paris"} JSON, API, MongoDB
Ensemble (unique) set {"python","data"} Déduplication (emails, tags)
Binaire bytes b"abc" Fichiers binaires, images, réseau

Exemples de variables

age = 30              # int
pi = 3.14             # float
nom = "Alice"         # str
est_data_engineer = True  # bool

print(age, type(age))
print(pi, type(pi))
print(nom, type(nom))
print(est_data_engineer, type(est_data_engineer))

Conversion de types

Il est fréquent de convertir des chaînes en nombres, par exemple après lecture d’un fichier.

age_str = "25"
age_int = int(age_str)

⚠️ Erreurs fréquentes : - Essayer de convertir une chaîne non numérique : int("abc") → ValueError ; - Additionner directement un str et un int : - "25" + 3 ❌ - int("25") + 3 ✔️

2. Conditions et Boucles

Les conditions permettent de prendre des décisions dans le code.
Les boucles permettent de répéter des actions automatiquement.

💡 En Data Engineering, ces structures sont essentielles pour : - Filtrer des données (conditions) - Traiter plusieurs fichiers (boucles) - Valider des règles métier (conditions) - Parcourir des bases de données (boucles)

2.1 Conditions (if / elif / else)

Les conditions testent des critères et exécutent du code selon le résultat.

Syntaxe de base

age = 20

if age < 18:
    print("Mineur")
elif age == 18:
    print("Tout juste majeur")
else:
    print("Majeur")

Fonctionnement :

  1. if : Première condition testée
  2. elif : Condition alternative (si if est fausse)
  3. else : Cas par défaut (si toutes les conditions sont fausses)

Opérateurs de comparaison

Opérateur Signification Exemple
== Égal à age == 18
!= Différent de age != 18
< Inférieur à age < 18
<= Inférieur ou égal age <= 18
> Supérieur à age > 18
>= Supérieur ou égal age >= 18

Conditions multiples

age = 25
pays = "France"

# Opérateur AND (et)
if age >= 18 and pays == "France":
    print("Peut voter en France")

# Opérateur OR (ou)
if age < 18 or age > 65:
    print("Tarif réduit")

# Opérateur NOT (négation)
if not (age < 18):
    print("Majeur")

2.2 Boucles (for et while)

Les boucles permettent d’exécuter du code de manière répétée.

Boucle for : Parcourir une séquence

# Parcourir une liste
noms = ["Alice", "Bob", "Charlie"]

for nom in noms:
    print(f"Bonjour {nom}")

# Résultat :
# Bonjour Alice
# Bonjour Bob
# Bonjour Charlie

Boucle avec range()

# range(n) génère les nombres de 0 à n-1
for i in range(5):
    print(f"Itération {i}")

# Résultat : 0, 1, 2, 3, 4

# range(début, fin, pas)
for i in range(0, 10, 2):
    print(i)  # 0, 2, 4, 6, 8

Boucle while : Répéter tant qu’une condition est vraie

compteur = 0

while compteur < 3:
    print(f"Compteur = {compteur}")
    compteur += 1  # ⚠️ IMPORTANT : incrémentation obligatoire

# Résultat :
# Compteur = 0
# Compteur = 1
# Compteur = 2

Contrôle de flux : break et continue

# break : Sortir immédiatement de la boucle
for i in range(10):
    if i == 5:
        break  # Arrête la boucle à 5
    print(i)  # Affiche 0, 1, 2, 3, 4

# continue : Passer à l'itération suivante
for i in range(5):
    if i == 2:
        continue  # Saute l'itération quand i=2
    print(i)  # Affiche 0, 1, 3, 4

Cas d’usage Data Engineering

# Exemple 1 : Traitement de fichiers multiples
fichiers = ["users_2024_01.csv", "users_2024_02.csv", "users_2024_03.csv"]

for fichier in fichiers:
    print(f"Traitement de {fichier}...")
    # Ici : logique de lecture/transformation
    # df = pd.read_csv(fichier)
    # process(df)

# Exemple 2 : Nettoyage de données
posts = [
    {"text": "Hello", "likes": 10},
    {"text": "", "likes": 5},       # ⚠️ Texte vide
    {"text": "Python", "likes": 20}
]

posts_valides = []

for post in posts:
    # Skip les posts vides
    if not post["text"].strip():
        continue
    
    # Nettoyer et garder
    post["text"] = post["text"].strip().lower()
    posts_valides.append(post)

print(posts_valides)
# [{'text': 'hello', 'likes': 10}, {'text': 'python', 'likes': 20}]

# Exemple 3 : Retry logic (tentatives multiples)
max_tentatives = 3
tentative = 0
succes = False

while tentative < max_tentatives and not succes:
    print(f"Tentative {tentative + 1}...")
    
    # Simulation d'une connexion
    # succes = tenter_connexion()
    
    tentative += 1
    
    if not succes and tentative < max_tentatives:
        print("Échec, nouvelle tentative...")

2.3 Boucles avancées : enumerate() et zip()

enumerate() : Obtenir l’index ET la valeur

fruits = ["pomme", "banane", "orange"]

# Sans enumerate (moins pratique)
for i in range(len(fruits)):
    print(f"{i}: {fruits[i]}")

# Avec enumerate (recommandé)
for index, fruit in enumerate(fruits):
    print(f"{index}: {fruit}")

# Avec enumerate démarrant à 1
for num, fruit in enumerate(fruits, start=1):
    print(f"Fruit #{num}: {fruit}")

zip() : Parcourir plusieurs listes simultanément

noms = ["Alice", "Bob", "Charlie"]
ages = [25, 30, 35]
villes = ["Paris", "Lyon", "Marseille"]

for nom, age, ville in zip(noms, ages, villes):
    print(f"{nom} a {age} ans et habite à {ville}")

# Résultat :
# Alice a 25 ans et habite à Paris
# Bob a 30 ans et habite à Lyon
# Charlie a 35 ans et habite à Marseille

Erreurs fréquentes et bonnes pratiques

❌ Erreur ✅ Correction 💡 Explication
Oublier : après if/for/while if age > 18: Syntaxe obligatoire
Mauvaise indentation Utiliser 4 espaces Python est sensible à l’indentation
Boucle infinie while Toujours incrémenter compteur += 1
Modifier liste pendant for Créer nouvelle liste Évite comportements imprévisibles
= au lieu de == if age == 18: = assigne, == compare

Exemples d’erreurs :

# ❌ Erreur 1 : Oublier le :
if age > 18
    print("Majeur")  # SyntaxError

# ✅ Correction
if age > 18:
    print("Majeur")

# ❌ Erreur 2 : Boucle infinie
compteur = 0
while compteur < 5:
    print(compteur)
    # Oubli d'incrémenter → boucle infinie !

# ✅ Correction
compteur = 0
while compteur < 5:
    print(compteur)
    compteur += 1

# ❌ Erreur 3 : Modifier liste pendant boucle
nombres = [1, 2, 3, 4, 5]
for n in nombres:
    if n % 2 == 0:
        nombres.remove(n)  # ⚠️ Comportement imprévisible

# ✅ Correction : List comprehension
nombres = [1, 2, 3, 4, 5]
nombres_impairs = [n for n in nombres if n % 2 != 0]

Récapitulatif

Structure Usage Exemple
if/elif/else Prendre des décisions Validation, filtrage
for Parcourir séquences Traiter fichiers, lignes
while Répéter tant que… Retry logic, polling
break Sortir de boucle Arrêt prématuré
continue Sauter itération Skip données invalides
enumerate() Index + valeur Numérotation
zip() Combiner listes Joindre données parallèles

Points clés à retenir

  1. Conditions : Utilisent == pour comparer (pas =)
  2. Indentation : 4 espaces obligatoires après :
  3. while : Toujours prévoir une sortie de boucle
  4. for : Préférer enumerate() si besoin de l’index
  5. List comprehension : Alternative élégante aux boucles simples

3. Structures de données

Python propose plusieurs structures très utilisées en data engineering.

Structure Ordonné Modifiable Duplicats Accès principal
list ✔️ ✔️ ✔️ index (0, 1, 2…)
dict ✔️ (3.7+) ✔️ clés uniques clé ("nom")
tuple ✔️ ❌ ✔️ index
set ❌ ✔️ ❌ appartenance (in)

3.1 Listes (list)

Une liste est une séquence ordonnée et modifiable.

# Création d'une liste
nombres = [10, 20, 30, 40]

# Accès par index
print(nombres[0])   # 10
print(nombres[2])   # 30

# Ajout en fin de liste
nombres.append(50)
print("Après append :", nombres)

# Insertion à une position précise
nombres.insert(1, 15)
print("Après insert :", nombres)

# Modification d'un élément
nombres[0] = 5
print("Après modification :", nombres)

# Suppression par valeur
nombres.remove(30)
print("Après remove :", nombres)

# Suppression par index
del nombres[0]
print("Après del :", nombres)

Création dynamique de listes

On crée très souvent des listes à partir d’autres listes, avec une boucle ou une list comprehension.

nombres = [1, 2, 3, 4, 5, 6]

# Version avec boucle
pairs = []
for n in nombres:
    if n % 2 == 0:
        pairs.append(n)

print("Pairs (boucle) :", pairs)

# Version list comprehension
pairs2 = [n for n in nombres if n % 2 == 0]
print("Pairs (list comprehension) :", pairs2)

⚠️ Erreurs fréquentes avec les listes : - nombres[10] alors que la liste a moins d’éléments → IndexError ; - nombres.remove(999) alors que 999 n’est pas dans la liste → ValueError.

3.2 Dictionnaires (dict)

Un dictionnaire stocke des paires clé → valeur. C’est l’équivalent naturel des objets JSON, très utilisé pour les APIs et NoSQL.

utilisateur = {
    "id": 1,
    "nom": "Alice",
    "ville": "Abidjan"
}

# Accès à une valeur par clé
print(utilisateur["nom"])  # Alice

# Ajout / modification
utilisateur["age"] = 30
utilisateur["ville"] = "Bouaké"

# Suppression
del utilisateur["id"]

print(utilisateur)

Accès sécurisé avec .get()

Utiliser dict.get() permet d’éviter un KeyError si la clé n’existe pas.

print(utilisateur.get("email"))           # None
print(utilisateur.get("email", "Inconnu"))  # Inconnu

Création dynamique : comptage d’occurrences

noms = ["bob", "alice", "bob", "charlie", "alice"]
compte = {}

for n in noms:
    compte[n] = compte.get(n, 0) + 1

print(compte)  # {'bob': 2, 'alice': 2, 'charlie': 1}

⚠️ Erreurs fréquentes avec les dictionnaires : - utilisateur["email"] alors que la clé n’existe pas → KeyError ; - supposer qu’un dictionnaire est indexé comme une liste (utilisateur[0]).

3.3 Tuples (tuple)

Un tuple est comme une liste non modifiable (immutable). On l’utilise pour représenter des collections fixes de valeurs : coordonnées, dates, etc.

coord = (5.0, 10.0)
print(coord[0])  # 5.0
print(coord[1])  # 10.0

# coord[0] = 20.0  # ❌ TypeError : un tuple n'est pas modifiable

En data engineering, les tuples sont utiles pour : - retourner plusieurs valeurs depuis une fonction ; - représenter des clés composites (ex : (année, mois)).

3.4 Ensembles (set)

Un ensemble (set) contient des valeurs uniques, sans ordre garanti. Très utile pour dédupliquer une liste.

tags = {"python", "data", "python"}
print(tags)  # 'python' n'apparaît qu'une seule fois

# Ajout
tags.add("engineer")

# Suppression (erreur si absent)
tags.remove("data")

# Suppression sans erreur si absent
tags.discard("ai")

print(tags)

💡 Exemple de déduplication :

emails = ["a@test.com", "b@test.com", "a@test.com"]
emails_uniques = list(set(emails))
print(emails_uniques)

⚠️ Erreurs fréquentes : - Compter sur l’ordre d’un set (l’ordre n’est pas garanti) ; - Utiliser remove() pour un élément possiblement absent → préférer discard().

3.5 Mini-exercice — Structures de données

Données

logs = [
    {"user": "alice", "status": 200},
    {"user": "bob", "status": 500},
    {"user": "alice", "status": 404},
    {"user": "bob", "status": 200},
]

Objectifs 1. Créer la liste de tous les utilisateurs (list) 2. Créer la liste unique des utilisateurs (set) 3. Créer un dictionnaire qui compte le nombre de requêtes par utilisateur (dict)

# À toi de jouer 😊

logs = [
    {"user": "alice", "status": 200},
    {"user": "bob", "status": 500},
    {"user": "alice", "status": 404},
    {"user": "bob", "status": 200},
]

utilisateurs = []            # TODO
utilisateurs_uniques = set() # TODO
compte_par_user = {}         # TODO

print(utilisateurs)
print(utilisateurs_uniques)
print(compte_par_user)
💡 Correction (cliquer pour afficher) 
logs = [
    {"user": "alice", "status": 200},
    {"user": "bob", "status": 500},
    {"user": "alice", "status": 404},
    {"user": "bob", "status": 200},
]

utilisateurs = []
utilisateurs_uniques = set()
compte_par_user = {}

for log in logs:
    user = log["user"]
    utilisateurs.append(user)
    utilisateurs_uniques.add(user)
    compte_par_user[user] = compte_par_user.get(user, 0) + 1

print(utilisateurs)
print(utilisateurs_uniques)
print(compte_par_user)

Comprehensions — Syntaxe puissante pour transformer les données

Les comprehensions sont une syntaxe Python élégante et performante pour créer des listes, dictionnaires ou sets en une seule ligne. C’est fondamental en Data Engineering pour transformer des données efficacement.

List Comprehension

Syntaxe : [expression for item in iterable if condition]

# ❌ Méthode classique (verbose)
nombres = [1, 2, 3, 4, 5]
carres = []
for n in nombres:
    carres.append(n ** 2)
print(carres)  # [1, 4, 9, 16, 25]

# ✅ List comprehension (recommandé)
carres = [n ** 2 for n in nombres]
print(carres)  # [1, 4, 9, 16, 25]

Avec condition (filtre)

# Garder seulement les nombres pairs
nombres = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
pairs = [n for n in nombres if n % 2 == 0]
print(pairs)  # [2, 4, 6, 8, 10]

# Exemple Data Engineering : filtrer les emails valides
emails = ["alice@test.com", "invalid", "bob@company.org", ""]
emails_valides = [e for e in emails if "@" in e and e]
print(emails_valides)  # ['alice@test.com', 'bob@company.org']

Cas d’usage Data Engineering

# Extraire les noms d'une liste de dictionnaires
users = [
    {"nom": "Alice", "actif": True},
    {"nom": "Bob", "actif": False},
    {"nom": "Charlie", "actif": True}
]

# Noms des utilisateurs actifs en majuscules
noms_actifs = [u["nom"].upper() for u in users if u["actif"]]
print(noms_actifs)  # ['ALICE', 'CHARLIE']

# Nettoyer une liste de fichiers
fichiers = ["data.csv", "readme.txt", "users.csv", "config.yaml"]
csv_files = [f for f in fichiers if f.endswith(".csv")]
print(csv_files)  # ['data.csv', 'users.csv']

Dict Comprehension

Syntaxe : {key: value for item in iterable if condition}

# Créer un dictionnaire à partir de deux listes
noms = ["alice", "bob", "charlie"]
ages = [25, 30, 35]

users_dict = {nom: age for nom, age in zip(noms, ages)}
print(users_dict)  # {'alice': 25, 'bob': 30, 'charlie': 35}

# Inverser un dictionnaire
original = {"a": 1, "b": 2, "c": 3}
inverse = {v: k for k, v in original.items()}
print(inverse)  # {1: 'a', 2: 'b', 3: 'c'}

Cas d’usage Data Engineering

# Transformer des données JSON
raw_data = [
    {"id": 1, "value": "100"},
    {"id": 2, "value": "200"},
    {"id": 3, "value": "300"}
]

# Créer un mapping id -> valeur (convertie en int)
id_to_value = {d["id"]: int(d["value"]) for d in raw_data}
print(id_to_value)  # {1: 100, 2: 200, 3: 300}

Set Comprehension

Syntaxe : {expression for item in iterable if condition}

# Valeurs uniques
nombres = [1, 2, 2, 3, 3, 3, 4, 4, 4, 4]
uniques = {n for n in nombres}
print(uniques)  # {1, 2, 3, 4}

# Domaines uniques des emails
emails = ["alice@gmail.com", "bob@yahoo.com", "charlie@gmail.com"]
domaines = {e.split("@")[1] for e in emails}
print(domaines)  # {'gmail.com', 'yahoo.com'}

Bonnes pratiques

✅ Faire ❌ Éviter
Comprehensions simples et lisibles Comprehensions imbriquées complexes
Une seule transformation Plusieurs opérations chaînées
Utiliser pour créer des collections Utiliser pour des effets de bord

Mini-exercice

Transformer cette liste de transactions :

transactions = [
    {"id": 1, "montant": 100, "devise": "EUR"},
    {"id": 2, "montant": 50, "devise": "USD"},
    {"id": 3, "montant": 200, "devise": "EUR"},
    {"id": 4, "montant": 75, "devise": "USD"}
]
  1. Créer une liste des montants en EUR uniquement
  2. Créer un dict {id: montant} pour les transactions > 60
💡 Solution 
# 1. Montants EUR
montants_eur = [t["montant"] for t in transactions if t["devise"] == "EUR"]
print(montants_eur)  # [100, 200]

# 2. Dict id -> montant (> 60)
id_montant = {t["id"]: t["montant"] for t in transactions if t["montant"] > 60}
print(id_montant)  # {1: 100, 3: 200, 4: 75}

4. Fonctions

Une fonction permet de :

  • réutiliser du code ;
  • encapsuler de la logique métier (nettoyage, validation…) ;
  • sécuriser la qualité de données (type, structure…).

Syntaxe générale :

def nom(param1, param2=valeur_par_defaut) -> type_retour:
    # traitement
    return resultat

Exemple :

def somme(a: int, b: int) -> int:
    """Retourne la somme de deux entiers."""
    return a + b

resultat = somme(3, 5)
print("Résultat :", resultat)

4.1 Fonction pure (sans effet externe)

def normaliser_nom(nom: str) -> str:
    """Nettoie un nom : supprime espaces, met en minuscule et capitalise."""
    return nom.strip().lower().capitalize()

print(normaliser_nom("  aLiCe  "))  # Alice

4.2 Paramètres + valeurs par défaut

def calculer_total(prix: float, quantite: int = 1, taxe: float = 0.18) -> float:
    """Retourne le prix total avec taxe."""
    return prix * quantite * (1 + taxe)

print(calculer_total(2000))  
print(calculer_total(2000, quantite=3, taxe=0.09))

4.3 Retourner plusieurs valeurs (tuple)

def stats_notes(notes: list[int]) -> tuple[float, float]:
    """Retourne moyenne et maximum d'une liste de notes."""
    moyenne = sum(notes) / len(notes)
    maxi = max(notes)
    return moyenne, maxi

m, mx = stats_notes([14, 9, 18])
print("Moyenne:", m, "Max:", mx)

4.4 Fonctions qui manipulent un dictionnaire

def extraire_champ(data: dict, champ: str, default=None):
    """Récupère un champ d'un dict, évite KeyError."""
    return data.get(champ, default)

user = {"nom": "Sara", "ville": "Paris"}
print(extraire_champ(user, "nom"))         # Sara
print(extraire_champ(user, "age", "N/A"))  # N/A

⚠️ Erreurs fréquentes : - Oublier les parenthèses lors de l’appel : somme au lieu de somme(3, 5) ; - Oublier return → la fonction retourne None ; - Ne pas respecter le nombre d’arguments attendus.

❌ Mauvaise pratique :

def ajouter(element, liste=[]):  # liste partagée !
    liste.append(element)
    return liste

✔️ Correct :

def ajouter(element, liste=None):
    if liste is None: liste = []
    liste.append(element)
    return liste

Mini-exercice — Fonctions sur des posts

Dataset simulé

posts = [
  {"user": "alice", "text": "  Hello World  "},
  {"user": "bob", "text": "Data Engineer ici"},
  {"user": "alice", "text": "Python est top "}
]

Instructions

Créer 3 fonctions :

  1. nettoyer_texte(text: str) -> str
    Nettoie le texte (supprime espaces, convertit en minuscules)

  2. longueur_post(post: dict) -> int
    Retourne la longueur du texte nettoyé d’un post

  3. stats_posts(posts: list[dict]) -> tuple[float, int, int]
    Retour attendu : (moyenne, maximum, minimum) des longueurs de texte

# À toi de jouer 😊

# TODO
💡 Correction (cliquer pour afficher) 
def nettoyer_texte(text: str) -> str:
    return text.strip().lower()

def longueur_post(post: dict) -> int:
    return len(nettoyer_texte(post["text"]))

def stats_posts(posts: list[dict]) -> tuple[float, int, int]:
    longueurs = [longueur_post(p) for p in posts]
    return (sum(longueurs)/len(longueurs), max(longueurs), min(longueurs))

print(stats_posts(posts))

Résultat attendu : (15.333333333333334, 19, 11)

4.5 *args et **kwargs — Fonctions flexibles

Ces syntaxes permettent de créer des fonctions qui acceptent un nombre variable d’arguments. Très utilisé pour créer des wrappers, des décorateurs, ou des fonctions génériques.

*args — Arguments positionnels variables

def somme(*args):
    """Accepte n'importe quel nombre d'arguments."""
    print(f"Arguments reçus : {args}")  # C'est un tuple
    return sum(args)

print(somme(1, 2))           # 3
print(somme(1, 2, 3, 4, 5))  # 15
print(somme())               # 0

**kwargs — Arguments nommés variables

def afficher_info(**kwargs):
    """Accepte n'importe quel argument nommé."""
    print(f"Arguments reçus : {kwargs}")  # C'est un dict
    for cle, valeur in kwargs.items():
        print(f"  {cle} = {valeur}")

afficher_info(nom="Alice", age=30, ville="Paris")
# Arguments reçus : {'nom': 'Alice', 'age': 30, 'ville': 'Paris'}
#   nom = Alice
#   age = 30
#   ville = Paris

Combiner les deux

def fonction_flexible(obligatoire, *args, option="defaut", **kwargs):
    """Ordre : obligatoire, *args, avec défaut, **kwargs"""
    print(f"Obligatoire : {obligatoire}")
    print(f"Args : {args}")
    print(f"Option : {option}")
    print(f"Kwargs : {kwargs}")

fonction_flexible("premier", 1, 2, 3, option="custom", extra="valeur")

Cas d’usage Data Engineering

def log_etl(etape: str, *messages, niveau: str = "INFO", **metadata):
    """Logger flexible pour pipeline ETL."""
    print(f"[{niveau}] {etape}")
    for msg in messages:
        print(f"  - {msg}")
    if metadata:
        print(f"  Metadata: {metadata}")

# Utilisation
log_etl(
    "EXTRACT",
    "Connexion établie",
    "1000 lignes lues",
    niveau="INFO",
    source="postgres",
    table="users"
)
# Wrapper pour ajouter du logging à n'importe quelle fonction
def avec_logging(func):
    def wrapper(*args, **kwargs):
        print(f"Appel de {func.__name__} avec args={args}, kwargs={kwargs}")
        result = func(*args, **kwargs)
        print(f"Résultat : {result}")
        return result
    return wrapper

5. Classes et Objets

Les classes permettent de créer des modèles (ou plans) d’objets combinant :

  • des données → attributs
  • des comportements → méthodes

En Data Engineering, elles sont très utiles pour structurer des pipelines, représenter des données complexes, encapsuler des transformations et rendre le code réutilisable.

5.1 Exemple simple : Modéliser un utilisateur 👤

class Utilisateur:
    def __init__(self, identifiant: int, nom: str, ville: str) -> None:
        self.identifiant = identifiant
        self.nom = nom
        self.ville = ville

    def presentation(self) -> str:
        return f"Je m'appelle {self.nom} et j'habite à {self.ville}."
u = Utilisateur(1, "Alice", "Abidjan")
print(u.presentation())

5.2 Ajouter une méthode utile

class Utilisateur:
    def __init__(self, identifiant: int, nom: str, ville: str) -> None:
        self.identifiant = identifiant
        self.nom = nom
        self.ville = ville

    def presentation(self) -> str:
        return f"Je m'appelle {self.nom} et j'habite à {self.ville}."

    def changer_ville(self, nouvelle_ville: str) -> None:
        self.ville = nouvelle_ville

5.3 Encapsulation : attributs protégés/privés

class Compte:
    def __init__(self, solde: float):
        self._solde = solde        # usage interne
        self.__secret = "XYZ123"   # privé via name mangling

5.4 Représentation textuelle (str)

class Utilisateur:
    def __init__(self, identifiant, nom, ville):
        self.identifiant = identifiant
        self.nom = nom
        self.ville = ville

    def __str__(self) -> str:
        return f"[{self.identifiant}] {self.nom} ({self.ville})"

5.5 Composition d’objets (objet dans un objet)

class Adresse:
    def __init__(self, rue: str, ville: str, pays: str) -> None:
        self.rue = rue
        self.ville = ville
        self.pays = pays

    def __str__(self) -> str:
        return f"{self.rue}, {self.ville} ({self.pays})"

class Utilisateur:
    def __init__(self, identifiant: int, nom: str, adresse: Adresse) -> None:
        self.identifiant = identifiant
        self.nom = nom
        self.adresse = adresse

    def presentation(self) -> str:
        return f"Je m'appelle {self.nom} et j'habite à {self.adresse}."

5.6 Pourquoi utiliser des classes en Data Engineering ?

class Transaction:
    def __init__(self, id, montant, devise, timestamp):
        self.id = id
        self.montant = montant
        self.devise = devise
        self.timestamp = timestamp

class Transaction:
    def __init__(self, id, montant, devise):
        self.id = id
        self.montant = montant
        self.devise = devise

    def montant_fcfa(self):
        taux = {"EUR": 655, "USD": 600}
        return self.montant * taux.get(self.devise, 1)

class ExtracteurCSV:
    def __init__(self, chemin):
        self.chemin = chemin

    def extract(self):
        with open(self.chemin) as f:
            return f.readlines()

event = EventHubRecord(payload)
event.clean()
event.validate()
event.to_parquet()

5.7 Erreurs fréquentes

  • Oublier self dans les méthodes.
  • Accéder à un attribut avant de l’avoir créé.
  • Utiliser une valeur mutable dans __init__ (list, dict).
  • Confondre composition et héritage.
  • Faire une classe « God Object » avec trop de responsabilités.

Dataclasses — Classes simplifiées pour les données

Les dataclasses (Python 3.7+) simplifient la création de classes qui servent principalement à stocker des données. C’est le standard moderne en Data Engineering pour modéliser des structures de données.

Problème avec les classes classiques

# ❌ Classe classique — verbose et répétitif
class User:
    def __init__(self, id: int, nom: str, email: str, actif: bool = True):
        self.id = id
        self.nom = nom
        self.email = email
        self.actif = actif
    
    def __repr__(self):
        return f"User(id={self.id}, nom='{self.nom}', email='{self.email}', actif={self.actif})"
    
    def __eq__(self, other):
        return self.id == other.id and self.nom == other.nom

✅ Solution avec dataclass

from dataclasses import dataclass

@dataclass
class User:
    id: int
    nom: str
    email: str
    actif: bool = True  # Valeur par défaut

# Utilisation
user = User(id=1, nom="Alice", email="alice@test.com")
print(user)  # User(id=1, nom='Alice', email='alice@test.com', actif=True)

# Comparaison automatique
user2 = User(id=1, nom="Alice", email="alice@test.com")
print(user == user2)  # True

Avantages des dataclasses

Avantage Description
__init__ auto-généré Plus besoin d’écrire le constructeur
__repr__ auto-généré Affichage lisible pour le debug
__eq__ auto-généré Comparaison par valeur
Type hints intégrés Documentation et validation IDE
Valeurs par défaut Syntaxe simple

Cas d’usage Data Engineering

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Transaction:
    id: int
    montant: float
    devise: str
    timestamp: str
    metadata: dict = field(default_factory=dict)  # Pour les types mutables

# Créer une transaction
tx = Transaction(
    id=1001,
    montant=150.50,
    devise="EUR",
    timestamp="2024-01-15T10:30:00Z"
)
print(tx)

# Accéder aux attributs
print(f"Montant: {tx.montant} {tx.devise}")

Dataclass immuable (frozen)

@dataclass(frozen=True)  # Immuable comme un tuple
class Config:
    host: str
    port: int
    database: str

config = Config(host="localhost", port=5432, database="warehouse")
# config.port = 3306  # ❌ Erreur : FrozenInstanceError

Convertir en dict/tuple

from dataclasses import asdict, astuple

@dataclass
class User:
    id: int
    nom: str
    email: str

user = User(1, "Alice", "alice@test.com")

# Conversion en dict (utile pour JSON)
user_dict = asdict(user)
print(user_dict)  # {'id': 1, 'nom': 'Alice', 'email': 'alice@test.com'}

# Conversion en tuple
user_tuple = astuple(user)
print(user_tuple)  # (1, 'Alice', 'alice@test.com')

⚠️ Attention : valeurs par défaut mutables

# ❌ ERREUR : liste partagée entre instances
@dataclass
class BadExample:
    items: list = []  # ValueError!

# ✅ CORRECT : utiliser field(default_factory=...)
from dataclasses import field

@dataclass
class GoodExample:
    items: list = field(default_factory=list)
    metadata: dict = field(default_factory=dict)

💡 Quand utiliser dataclass vs classe classique ?

Situation Recommandation
Stocker des données (models, records) ✅ @dataclass
Logique métier complexe Classe classique
Configuration, paramètres ✅ @dataclass(frozen=True)
Validation avancée Pydantic (module suivant)

6. Gestion d’erreurs — (Indispensable en Data Engineering)

En Data Engineering, les erreurs sont inévitables :

  • fichiers manquants
  • API indisponible
  • JSON mal formé
  • division par zéro
  • connexion BD échouée
  • typage incorrect

👉 La bonne pratique consiste à capturer, expliquer, puis continuer proprement.
C’est le rôle de try / except.

🎯 Exemple simple — division sécurisée

def division(a: float, b: float) -> float | None:
    try:
        return a / b
    except ZeroDivisionError:
        print("❌ Erreur : division par zéro.")
        return None
print(division(10, 2))  # OK
print(division(10, 0))  # Erreur gérée

Exemple plus réaliste — lecture de fichier

def lire_csv(chemin: str) -> list | None:
    try:
        with open(chemin, "r") as f:
            return f.readlines()
    except FileNotFoundError:
        print(f"❌ Fichier introuvable : {chemin}")
        return None

💡 else et finally

On peut améliorer la lisibilité avec else et finally :

try:
    result = 10 / 2
except ZeroDivisionError:
    print("Erreur")
else:
    print("Aucune erreur, résultat =", result)
finally:
    print("Bloc exécuté dans tous les cas")

Exemple Data Engineering : appel API sécurisé

import requests

def fetch_json(url: str) -> dict | None:
    try:
        response = requests.get(url, timeout=3)
        response.raise_for_status()   # Génère une erreur HTTP si code ≠ 200
        return response.json()
    except requests.exceptions.HTTPError as e:
        print("❌ Erreur HTTP :", e)
    except requests.exceptions.Timeout:
        print("⏱️ Timeout : serveur trop lent")
    except ValueError:
        print("❌ JSON mal formé")
    except Exception as e:
        print("⚠️ Erreur inconnue :", e)
    return None

⚠️ Erreurs fréquentes à éviter

❌ Mauvaise pratique ✅ Bonne pratique
except: (attrape tout) Spécifier l’erreur : except ValueError:
Cacher l’erreur sans message Fournir un contexte 🗃️
Retourner n’importe quoi Retour cohérent (None ou valeur par défaut)
Mettre trop de logique dans try Limiter au strict nécessaire
Ignorer les erreurs silencieusement Logguer ou notifier

Conseil pro (très utile en Data Engineering)

Utiliser raise pour propager l’erreur si elle doit être traitée ailleurs :

def parse_json(data: str) -> dict:
    try:
        return json.loads(data)
    except json.JSONDecodeError as e:
        raise ValueError(f"JSON invalide : {e}")

À retenir

  1. Toujours attraper le type exact d’erreur.
  2. Toujours expliquer l’erreur (message clair).
  3. Toujours garder un comportement cohérent (retour None ou valeur défaut).
  4. Les erreurs silencieuses sont pires que les erreurs visibles.
  5. Les pipelines cassent souvent — gérer les erreurs = être un vrai Data Engineer.

7. Modules et Imports — Structurer son code comme un Data Engineer

En Python :

  • Un module = un fichier .py
  • Un package = un dossier contenant plusieurs modules + un fichier __init__.py

👉 Cela permet d’organiser un projet data en blocs logiques :
ingestion, nettoyage, transformation, validation, etc.

Exemple de structure de projet (propre & professionnelle)

project/
├── utils/               ← Package (outils réutilisables)
│   ├── __init__.py      ← Indique que `utils` est un package
│   └── math_utils.py     ← Module
├── processors/           ← Package pour le traitement
│   ├── __init__.py
│   └── text_cleaner.py   ← Module
└── main.py               ← Point d’entrée du projet

Contenu du module : utils/math_utils.py

def somme(a, b):
    return a + b

Contenu d’un autre module : processors/text_cleaner.py

def nettoyer_texte(text: str) -> str:
    return text.strip().lower()

Importer dans main.py

from utils.math_utils import somme
from processors.text_cleaner import nettoyer_texte

print(somme(2, 3))
print(nettoyer_texte("  Hello WORLD  "))

Import absolu vs import relatif

Import absolu (recommandé)

from utils.math_utils import somme

➡️ Le plus lisible, idéal pour projets pro / Data Engineering.

Import relatif (utile dans les packages)

from .math_utils import somme

➡️ Courant dans les gros projets et dans les packages pip.

Pourquoi c’est important en Data Engineering ?

  • Créer des modules = rendre ton code réutilisable (API, pipelines, notebooks)
  • Structurer ton projet = éviter le “script spaghetti”
  • Faciliter les tests unitaires
  • Faciliter la maintenance d’un pipeline data
  • Réduire les erreurs de duplication de logique

⚠️ Erreurs fréquentes & solutions

❌ Problème 💡 Cause ✅ Solution
ModuleNotFoundError exécuter Python depuis le mauvais dossier Toujours lancer Python depuis la racine du projet
Import relatif impossible absence du fichier __init__.py Ajouter __init__.py dans le dossier
Modules dupliqués fichiers ayant le même nom dans deux dossiers Renommer ou structurer les packages
import * imports imprécis Toujours importer explicitement

Astuce pro : vérifier ton PYTHONPATH

import sys
print(sys.path)

Cela indique où Python cherche les modules.

Rappel essentiel

Importer = Réutiliser.
Structurer = Devenir un vrai Data Engineer.

8. Manipulation de fichiers — Compétence essentielle du Data Engineer

Dans un pipeline Data, on manipule en permanence des fichiers :

  • fichiers texte (logs, outputs)
  • fichiers CSV (exports métier, ingestion)
  • fichiers JSON (APIs, NoSQL, événements Kafka)
  • dossiers qui contiennent les données

Python fournit des outils natifs très puissants pour cela :

  • pathlib.Path → gérer les chemins et dossiers
  • open() → lire/écrire des fichiers texte
  • json → sérialiser/désérialiser des données JSON

Initialisation du dossier data/

from pathlib import Path
import json

# Création du dossier de travail
DATA_DIR = Path("data")
DATA_DIR.mkdir(exist_ok=True)

8.1 Manipulation d’un fichier texte (logs, outputs…)

texte_path = DATA_DIR / "exemple.txt"

# Écriture
with open(texte_path, "w", encoding="utf-8") as f:
    f.write("Bonjour Data Engineer\n")

# Lecture
with open(texte_path, "r", encoding="utf-8") as f:
    contenu = f.read()

print("Contenu du fichier :", contenu)

Bonnes pratiques :

  • Toujours utiliser encoding="utf-8"
  • Toujours utiliser with ... (fermeture automatique du fichier)

8.2 Manipulation d’un fichier JSON

Format le plus utilisé dans : - APIs REST - MongoDB - Events Kafka / Kinesis - Configurations de job

json_path = DATA_DIR / "utilisateur.json"
utilisateur = {"nom": "Alice", "age": 30}

# Écriture JSON
with open(json_path, "w", encoding="utf-8") as f:
    json.dump(utilisateur, f, ensure_ascii=False, indent=2)

# Lecture JSON
with open(json_path, "r", encoding="utf-8") as f:
    utilisateur_charge = json.load(f)

print("Utilisateur chargé :", utilisateur_charge)

💡 ensure_ascii=False permet d’écrire proprement les accents.
💡 indent=2 rend le JSON lisible (log, debug, audits…).

8.3 (Optionnel) Manipulation d’un CSV avec Pandas

Utilisé en ingestion de données. On en reparlera plus au module 5.

import pandas as pd

csv_path = DATA_DIR / "exemple.csv"

df = pd.DataFrame({
    "nom": ["Alice", "Bob"],
    "age": [30, 25]
})

df.to_csv(csv_path, index=False)

df_loaded = pd.read_csv(csv_path)
print(df_loaded)

⚠️ Erreurs fréquentes & comment les éviter

❌ Problème 💡 Cause ✅ Solution
FileNotFoundError Mauvais chemin Toujours utiliser Path() et vérifier path.exists()
Accents cassés (é, à…) Mauvais encoding Toujours encoding="utf-8"
Fichier non fermé open() sans contexte Toujours utiliser with open(...)
JSON mal formé écrit à la main Toujours utiliser json.dump / json.load
Chemins relatifs non fiables mauvais dossier d’exécution Utiliser Path(__file__).resolve().parent dans un projet réel

Points clés à retenir

  1. pathlib.Path simplifie la gestion des chemins.
  2. with open(...) est obligatoire pour éviter les fuites de fichier.
  3. JSON = format standard du Data Engineering (MongoDB, API, logs…).
  4. Toujours contrôler l’encoding lors de la lecture/écriture.
  5. Le dossier data/ centralise vos fichiers dans un projet propre.

9. Packages et pip — Gérer les dépendances comme un Data Engineer

Python devient puissant grâce à ses packages externes :
Pandas, Requests, SQLAlchemy, PyMongo, Polars, FastAPI, etc.

Pour installer ces packages, on utilise pip, le gestionnaire officiel de Python.

Installer un package avec pip

Exemple : installer requests pour faire des appels HTTP (API REST).

python -m pip install requests

👉 Pourquoi python -m pip et pas juste pip install ?
Parce que cela garantit qu’on utilise le pip lié à la bonne version de Python, surtout si plusieurs versions sont installées.

Utilisation dans un script

import requests

response = requests.get("https://api.github.com")
print(response.status_code)

Installer plusieurs packages — requirements.txt

Dans un vrai projet Data Engineering, on liste les dépendances dans un fichier :

requests
pandas
sqlalchemy
pymongo

Puis on installe tout d’un coup :

python -m pip install -r requirements.txt

Mettre à jour un package

python -m pip install --upgrade requests

Désinstaller un package

python -m pip uninstall requests

Où sont installés les packages ?

python -m pip show requests

Donne : version, emplacement, dépendances, etc.

⚠️ Erreurs fréquentes & bonnes pratiques

❌ Problème 💡 Cause ✅ Solution
pip: command not found Python mal installé Utiliser python -m pip
Installer dans le mauvais Python Plusieurs versions installées Toujours python -m pip install
Version incompatible Conflit de dépendances Spécifier une version : requests==2.31.0
Installer globalement Risque de casser le système Utiliser un venv (python -m venv .venv)

⭐ Astuce Pro : figer les versions pour la production

python -m pip freeze > requirements.txt

➡️ Cela capture exactement les versions utilisées dans ton environnement.

Générateurs (yield) — Traiter de gros volumes sans saturer la mémoire

Les générateurs produisent des valeurs une par une, sans tout charger en mémoire. C’est essentiel en Data Engineering pour traiter des fichiers volumineux ou des flux de données.

Problème : charger tout en mémoire

# ❌ Charge TOUT le fichier en mémoire
def lire_fichier_complet(chemin):
    with open(chemin) as f:
        return f.readlines()  # Liste de TOUTES les lignes

# Si le fichier fait 10 Go... 💥 MemoryError
lignes = lire_fichier_complet("huge_file.csv")

✅ Solution : générateur avec yield

# ✅ Lit UNE ligne à la fois
def lire_fichier_ligne_par_ligne(chemin):
    with open(chemin) as f:
        for ligne in f:
            yield ligne.strip()  # Retourne et "pause"

# Utilisation : ne charge qu'une ligne à la fois
for ligne in lire_fichier_ligne_par_ligne("huge_file.csv"):
    process(ligne)  # Traite ligne par ligne

Différence return vs yield

return yield
Retourne une valeur et termine Retourne une valeur et pause
Fonction normale Générateur
Tout en mémoire Une valeur à la fois 
# Fonction normale
def carres_liste(n):
    result = []
    for i in range(n):
        result.append(i ** 2)
    return result  # Retourne toute la liste

# Générateur
def carres_generateur(n):
    for i in range(n):
        yield i ** 2  # Retourne un par un

# Test mémoire
import sys
liste = carres_liste(1000000)
gen = carres_generateur(1000000)

print(sys.getsizeof(liste))  # ~8 Mo
print(sys.getsizeof(gen))    # ~200 octets !

Cas d’usage Data Engineering

Lire un CSV volumineux

def lire_csv_en_chunks(chemin, chunk_size=1000):
    """Lit un CSV par lots de chunk_size lignes."""
    with open(chemin) as f:
        header = next(f).strip().split(",")
        chunk = []
        
        for ligne in f:
            valeurs = ligne.strip().split(",")
            row = dict(zip(header, valeurs))
            chunk.append(row)
            
            if len(chunk) >= chunk_size:
                yield chunk
                chunk = []
        
        if chunk:  # Dernier lot
            yield chunk

# Utilisation
for batch in lire_csv_en_chunks("users.csv", chunk_size=500):
    print(f"Traitement de {len(batch)} lignes...")
    # insert_to_database(batch)

Pipeline de transformation

def extraire(source):
    for item in source:
        yield item

def transformer(items):
    for item in items:
        item["nom"] = item["nom"].upper()
        yield item

def filtrer(items, condition):
    for item in items:
        if condition(item):
            yield item

# Pipeline chaîné — évaluation paresseuse (lazy)
source = [{"nom": "alice"}, {"nom": "bob"}, {"nom": "charlie"}]
pipeline = filtrer(
    transformer(extraire(source)),
    lambda x: len(x["nom"]) > 3
)

for item in pipeline:
    print(item)  # {'nom': 'ALICE'}, {'nom': 'CHARLIE'}

Generator Expressions (syntaxe courte)

# List comprehension — crée une liste en mémoire
carres_liste = [x**2 for x in range(1000000)]

# Generator expression — crée un générateur
carres_gen = (x**2 for x in range(1000000))  # Parenthèses !

# Utile pour les agrégations
total = sum(x**2 for x in range(1000000))  # Efficient

Créer et exécuter un script Python (.py)

Jusqu’ici, tu as travaillé dans un Notebook Jupyter. Mais en Data Engineering, on écrit surtout des scripts .py qui s’exécutent depuis le terminal ou sont orchestrés par Airflow, cron, etc.

Notebook vs Script — Quand utiliser quoi ?

Critère Notebook (.ipynb) Script (.py)
Exploration ✅ Idéal ❌ Pas adapté
Visualisation ✅ Graphiques inline ⚠️ Doit sauvegarder
Production ❌ Difficile à orchestrer ✅ Standard
Tests unitaires ❌ Compliqué ✅ Facile avec pytest
Git / Code review ⚠️ Diffs illisibles ✅ Diffs propres
CI/CD ❌ Pas adapté ✅ Standard
Airflow / Orchestration ⚠️ Possible mais pas idéal ✅ Natif

💡 Règle : Notebook pour explorer, Script pour produire.

Créer ton premier script

  1. Ouvre VS Code
  2. Crée un nouveau fichier : File > New File
  3. Sauvegarde-le avec l’extension .py : mon_script.py

Exemple : hello.py

# hello.py
print("Hello, Data Engineer!")

▶️ Exécuter depuis le terminal

# Se placer dans le dossier du script
cd /chemin/vers/mon/projet

# Exécuter le script
python hello.py
# ou
python3 hello.py

Résultat :

Hello, Data Engineer!

Structure standard : if __name__ == "__main__"

C’est LA structure que tu verras dans tous les scripts Python professionnels.

# etl_simple.py

def extract():
    """Extrait les données."""
    print(" Extraction...")
    return [{"id": 1, "nom": "Alice"}, {"id": 2, "nom": "Bob"}]

def transform(data):
    """Transforme les données."""
    print(" Transformation...")
    return [{**d, "nom": d["nom"].upper()} for d in data]

def load(data):
    """Charge les données."""
    print(" Chargement...")
    for row in data:
        print(f"  Inséré : {row}")

def main():
    """Point d'entrée principal."""
    print(" Démarrage du pipeline ETL")
    
    data = extract()
    data = transform(data)
    load(data)
    
    print(" Pipeline terminé !")

# Ce bloc s'exécute SEULEMENT si le script est lancé directement
if __name__ == "__main__":
    main()

Pourquoi if __name__ == "__main__" ?

Situation __name__ vaut Le bloc s’exécute ?
python etl_simple.py "__main__" ✅ Oui
import etl_simple "etl_simple" ❌ Non

Cela permet de : - Réutiliser les fonctions dans d’autres scripts (import etl_simple) - Exécuter le script directement (python etl_simple.py)

Passer des arguments au script

Méthode simple : sys.argv

# script_args.py
import sys

def main():
    print(f"Arguments reçus : {sys.argv}")
    
    # sys.argv[0] = nom du script
    # sys.argv[1], sys.argv[2], ... = arguments
    
    if len(sys.argv) < 2:
        print("Usage : python script_args.py <fichier>")
        sys.exit(1)
    
    fichier = sys.argv[1]
    print(f"Traitement de : {fichier}")

if __name__ == "__main__":
    main()
python script_args.py data.csv
# Arguments reçus : ['script_args.py', 'data.csv']
# Traitement de : data.csv

Méthode pro : argparse

# etl_avec_args.py
import argparse

def main():
    # Créer le parser
    parser = argparse.ArgumentParser(
        description="Pipeline ETL simple"
    )
    
    # Définir les arguments
    parser.add_argument(
        "--input", "-i",
        required=True,
        help="Fichier d'entrée (CSV)"
    )
    parser.add_argument(
        "--output", "-o",
        default="output.csv",
        help="Fichier de sortie (défaut: output.csv)"
    )
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Mode verbeux"
    )
    
    # Parser les arguments
    args = parser.parse_args()
    
    # Utiliser les arguments
    print(f"Input  : {args.input}")
    print(f"Output : {args.output}")
    print(f"Verbose: {args.verbose}")

if __name__ == "__main__":
    main()
# Afficher l'aide
python etl_avec_args.py --help

# Exécuter avec arguments
python etl_avec_args.py --input data.csv --output result.csv --verbose
python etl_avec_args.py -i data.csv -o result.csv -v

Structure d’un projet Data Engineering

mon_projet/
├── src/
│   ├── __init__.py
│   ├── extract.py
│   ├── transform.py
│   └── load.py
├── scripts/
│   └── run_etl.py      ← Point d'entrée
├── tests/
│   └── test_transform.py
├── data/               ← ⚠️ Dans .gitignore
├── requirements.txt
├── .gitignore
└── README.md

Exercice pratique

Créer un script compter_lignes.py qui : 1. Prend un fichier en argument 2. Compte le nombre de lignes 3. Affiche le résultat

💡 Solution 
# compter_lignes.py
import argparse
from pathlib import Path

def compter_lignes(chemin: str) -> int:
    """Compte les lignes d'un fichier."""
    path = Path(chemin)
    if not path.exists():
        raise FileNotFoundError(f"Fichier non trouvé : {chemin}")
    
    with open(path, "r", encoding="utf-8") as f:
        return sum(1 for _ in f)

def main():
    parser = argparse.ArgumentParser(description="Compte les lignes d'un fichier")
    parser.add_argument("fichier", help="Chemin du fichier à analyser")
    args = parser.parse_args()
    
    try:
        nb_lignes = compter_lignes(args.fichier)
        print(f"Le fichier '{args.fichier}' contient {nb_lignes} lignes.")
    except FileNotFoundError as e:
        print(f"Erreur : {e}")
        exit(1)

if __name__ == "__main__":
    main()
python compter_lignes.py mon_fichier.csv

Context Managers — Gérer les ressources proprement

Tu as déjà utilisé with open(...) pour les fichiers. C’est un context manager ! Ce pattern garantit que les ressources sont toujours libérées, même en cas d’erreur.

Le problème sans context manager

# ❌ Risque : fichier non fermé si erreur
f = open("data.txt", "r")
data = f.read()
# Si une erreur se produit ici...
process(data)  # Le fichier reste ouvert !
f.close()
# ✅ Avec context manager : fermeture garantie
with open("data.txt", "r") as f:
    data = f.read()
    process(data)
# Fichier automatiquement fermé, même en cas d'erreur

Cas d’usage Data Engineering

Ressource Pourquoi un context manager
Fichiers Fermer après lecture/écriture
Connexions DB Libérer la connexion
Transactions Commit ou rollback
Fichiers temporaires Supprimer après usage
Verrous (locks) Libérer le verrou

Créer son propre context manager

Méthode 1 : avec une classe

class DatabaseConnection:
    def __init__(self, host: str, database: str):
        self.host = host
        self.database = database
        self.connection = None
    
    def __enter__(self):
        """Appelé au début du bloc 'with'."""
        print(f"🔌 Connexion à {self.host}/{self.database}...")
        self.connection = f"Connection to {self.database}"  # Simule une connexion
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Appelé à la fin du bloc 'with' (même si erreur)."""
        print(f"🔌 Fermeture de la connexion...")
        self.connection = None
        # Retourner False pour propager les exceptions
        return False

# Utilisation
with DatabaseConnection("localhost", "warehouse") as conn:
    print(f"Connexion active : {conn}")
    # Faire des requêtes...
# Connexion automatiquement fermée ici

Méthode 2 : avec @contextmanager (plus simple)

from contextlib import contextmanager

@contextmanager
def timer(nom: str):
    """Mesure le temps d'exécution d'un bloc."""
    import time
    start = time.time()
    print(f" Début : {nom}")
    
    yield  # Le bloc 'with' s'exécute ici
    
    elapsed = time.time() - start
    print(f" Fin : {nom} ({elapsed:.2f}s)")

# Utilisation
with timer("Traitement ETL"):
    # Simuler un traitement long
    import time
    time.sleep(1)
    print("Traitement en cours...")

# Output:
# Début : Traitement ETL
# Traitement en cours...
# Fin : Traitement ETL (1.00s)

Exemple complet : Transaction DB

from contextlib import contextmanager

@contextmanager
def transaction(connection):
    """Gère une transaction avec commit/rollback automatique."""
    try:
        print(" Début de la transaction")
        yield connection
        print(" COMMIT")
        # connection.commit()
    except Exception as e:
        print(f" ROLLBACK : {e}")
        # connection.rollback()
        raise

# Utilisation
with transaction("ma_connexion") as conn:
    print("Insertion de données...")
    # Si une erreur ici → rollback automatique

💡 Context managers utiles de la stdlib

from contextlib import suppress, redirect_stdout
import tempfile

# Ignorer silencieusement une erreur
with suppress(FileNotFoundError):
    os.remove("fichier_peut_etre_absent.txt")

# Fichier temporaire (supprimé automatiquement)
with tempfile.NamedTemporaryFile(mode="w", delete=True) as tmp:
    tmp.write("données temporaires")
    print(f"Fichier temp : {tmp.name}")
# Fichier supprimé ici

10. Logging — Traçabilité indispensable en Data Engineering

Dans un vrai pipeline Data (ingestion, nettoyage, transformation…), on doit suivre ce qu’il se passe :

  • Fichier introuvable ?
  • API trop lente ?
  • Format JSON invalide ?
  • Données anormales ?
  • ETL en retard ?

➡️ print() ne suffit pas, car il ne permet ni filtrage, ni niveaux, ni logs dans un fichier.
➡️ Le module logging est le standard utilisé en entreprise.

10.1 Pourquoi utiliser logging ?

Avantage Description
Niveaux de log DEBUG, INFO, WARNING, ERROR, CRITICAL
Filtrage des messages On peut afficher seulement WARNING+
Logs dans un fichier Indispensable en production
Standard Python Compatible Airflow, FastAPI, ETL, microservices
Thread-safe Fonctionne même avec du multithreading

🔧 Configuration minimale recommandée

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)

Explications : - level=INFO → DEBUG est ignoré - %(asctime)s → timestamp (important dans les pipelines) - %(levelname)s → niveau (INFO, ERROR…) - %(message)s → contenu du message

Exemple d’utilisation

logging.debug("Message DEBUG (non affiché en mode INFO)")
logging.info("Démarrage du mini-script")
logging.warning("Attention : données manquantes")
logging.error("Erreur : échec de connexion API")
logging.critical("CRITIQUE : le pipeline doit s'arrêter !")

10.2 Écrire les logs dans un fichier (cas réel Data Engineering)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    filename="pipeline.log",
    filemode="a",  # append au lieu de réécrire
)

➡️ Très utilisé dans : - scripts Airflow - traitements batch - pipelines de production

10.3 Exemple typique dans une fonction

def charger_json(chemin: str) -> dict | None:
    logging.info(f"Chargement du fichier : {chemin}")
    try:
        with open(chemin, "r", encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        logging.error(f"Fichier introuvable : {chemin}")
    except json.JSONDecodeError:
        logging.error("JSON invalide")
    return None

⚠️ Erreurs fréquentes & comment les éviter

❌ Mauvaise pratique 💡 Pourquoi ✅ Bonne pratique
Utiliser print() partout impossible à filtrer/logguer Utiliser logging.info()
Appeler logging.basicConfig() plusieurs fois ne fonctionne que la 1ère fois Configurer un seul logger global
Logger des données sensibles fuite de secrets/mots de passe Filtrer/masquer les champs sensibles
Logs trop verbeux ralentissent les pipelines Utiliser DEBUG seulement en dev
Logs insuffisants difficile de diagnostiquer Logger les erreurs + contexte

⭐ Astuce pro : Logger dans la console et dans un fichier

logger = logging.getLogger("pipeline")
logger.setLevel(logging.INFO)

# Handler console
console = logging.StreamHandler()
console.setLevel(logging.INFO)

# Handler fichier
file = logging.FileHandler("pipeline.log")
file.setLevel(logging.INFO)

# Format
fmt = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
console.setFormatter(fmt)
file.setFormatter(fmt)

# Ajouter handlers
logger.addHandler(console)
logger.addHandler(file)

logger.info("Pipeline démarré")

À retenir

  1. logging = indispensable en Data Engineering
  2. Toujours configurer : niveau + format
  3. Utiliser un fichier log en production
  4. Jamais de print() dans un vrai pipeline
  5. Bien choisir le niveau (INFO, WARNING, ERROR…)

11. Exercices pratiques

Essaie de résoudre ces exercices avant d’afficher les corrections.

Exercice 1 – Validation d’âge

Écrire une fonction est_majeur(age: int) -> bool qui : - retourne True si age est supérieur ou égal à 18 ; - sinon retourne False.

# À toi de jouer
def est_majeur(age: int) -> bool:
    # TODO: compléter
    pass

print(est_majeur(15))  # attendu: False
print(est_majeur(18))  # attendu: True
Afficher une solution possible 
def est_majeur(age: int) -> bool:
    return age >= 18

print(est_majeur(15))  # False
print(est_majeur(18))  # True

Exercice 2 – Compter les posts par utilisateur

On dispose d’une liste de dictionnaires représentant des posts :

posts = [
    {"user": "alice", "text": "Hello"},
    {"user": "bob", "text": "Salut"},
    {"user": "alice", "text": "Rebonjour"},
]

Écrire une fonction compter_posts_par_utilisateur(posts) qui retourne :

{"alice": 2, "bob": 1}
# À toi de jouer
posts = [
    {"user": "alice", "text": "Hello"},
    {"user": "bob", "text": "Salut"},
    {"user": "alice", "text": "Rebonjour"},
]

def compter_posts_par_utilisateur(posts: list[dict]) -> dict:
    # TODO: compléter
    result = {}
    return result

print(compter_posts_par_utilisateur(posts))
Afficher une solution possible 
def compter_posts_par_utilisateur(posts: list[dict]) -> dict:
    result = {}
    for p in posts:
        user = p["user"]
        if user not in result:
            result[user] = 0
        result[user] += 1
    return result

print(compter_posts_par_utilisateur(posts))  # {'alice': 2, 'bob': 1}

Exercice 3 – Classe Post

Créer une classe Post avec : - attributs : auteur (str), texte (str) ; - méthode longueur() qui retourne la longueur du texte.

# À toi de jouer
class Post:
    def __init__(self, auteur: str, texte: str) -> None:
        # TODO: stocker les attributs
        pass

    def longueur(self) -> int:
        # TODO: retourner la longueur du texte
        return 0

p = Post("alice", "Bonjour tout le monde")
print(p.longueur())  # attendu: longueur de la phrase
Afficher une solution possible 
class Post:
    def __init__(self, auteur: str, texte: str) -> None:
        self.auteur = auteur
        self.texte = texte

    def longueur(self) -> int:
        return len(self.texte)

p = Post("alice", "Bonjour tout le monde")
print(p.longueur())

Introduction aux tests avec pytest

Tester son code est essentiel en Data Engineering. Un bug dans un pipeline peut corrompre des millions de lignes de données ! pytest est le framework de test standard en Python.

Installation

pip install pytest

Premier test avec assert

assert vérifie qu’une condition est vraie. Si elle est fausse → erreur.

# test_basics.py

def addition(a, b):
    return a + b

def test_addition():
    """Test simple avec assert."""
    assert addition(2, 3) == 5
    assert addition(0, 0) == 0
    assert addition(-1, 1) == 0

def test_addition_floats():
    """Test avec des floats."""
    result = addition(0.1, 0.2)
    assert abs(result - 0.3) < 0.0001  # Comparaison floats
# Exécuter les tests
pytest test_basics.py

# Avec plus de détails
pytest test_basics.py -v

Tester des fonctions Data Engineering

# src/transform.py
def nettoyer_email(email: str) -> str:
    """Nettoie un email : minuscules, strip."""
    if not email or "@" not in email:
        raise ValueError("Email invalide")
    return email.strip().lower()

def filtrer_actifs(users: list[dict]) -> list[dict]:
    """Garde uniquement les users actifs."""
    return [u for u in users if u.get("actif", False)]
# tests/test_transform.py
import pytest
from src.transform import nettoyer_email, filtrer_actifs

class TestNettoyerEmail:
    """Tests pour nettoyer_email."""
    
    def test_email_valide(self):
        assert nettoyer_email("Alice@Test.COM") == "alice@test.com"
    
    def test_email_avec_espaces(self):
        assert nettoyer_email("  bob@test.com  ") == "bob@test.com"
    
    def test_email_invalide_sans_arobase(self):
        with pytest.raises(ValueError):
            nettoyer_email("invalid_email")
    
    def test_email_vide(self):
        with pytest.raises(ValueError):
            nettoyer_email("")


class TestFiltrerActifs:
    """Tests pour filtrer_actifs."""
    
    def test_filtre_users_actifs(self):
        users = [
            {"nom": "Alice", "actif": True},
            {"nom": "Bob", "actif": False},
            {"nom": "Charlie", "actif": True}
        ]
        result = filtrer_actifs(users)
        assert len(result) == 2
        assert all(u["actif"] for u in result)
    
    def test_liste_vide(self):
        assert filtrer_actifs([]) == []
    
    def test_tous_inactifs(self):
        users = [{"nom": "X", "actif": False}]
        assert filtrer_actifs(users) == []

Exécuter les tests

# Tous les tests du dossier
pytest

# Un fichier spécifique
pytest tests/test_transform.py

# Une classe spécifique
pytest tests/test_transform.py::TestNettoyerEmail

# Un test spécifique
pytest tests/test_transform.py::TestNettoyerEmail::test_email_valide

# Avec couverture de code
pip install pytest-cov
pytest --cov=src

Structure recommandée

mon_projet/
├── src/
│   ├── __init__.py
│   ├── extract.py
│   └── transform.py
├── tests/
│   ├── __init__.py
│   ├── test_extract.py
│   └── test_transform.py
├── pytest.ini          # Configuration pytest (optionnel)
└── requirements.txt

💡 Bonnes pratiques

Pratique Explication
Nommer test_*.py pytest les détecte automatiquement
Un assert par test Plus facile à débuguer
Tester les cas limites Listes vides, None, valeurs négatives
Tester les erreurs pytest.raises(Exception)
Exécuter avant chaque commit Évite les régressions

12. Quiz final

Teste tes connaissances ! Réponds mentalement puis vérifie les réponses.

❓ Q1. Quel type correspond à une chaîne de caractères ?

  1. str
  2. text
  3. char
💡 Voir la réponse

✅ Réponse : a — str est le type pour les chaînes de caractères.

❓ Q2. Quelle structure est la plus adaptée pour représenter un objet JSON ?

  1. list
  2. dict
  3. tuple
💡 Voir la réponse

✅ Réponse : b — dict avec des paires clé/valeur, comme JSON.

❓ Q3. Laquelle de ces boucles risque le plus de devenir infinie ?

  1. for
  2. while
💡 Voir la réponse

✅ Réponse : b — while continue tant que la condition est vraie.

❓ Q4. Quel mot-clé permet de gérer une erreur ?

  1. error
  2. except
  3. catch
💡 Voir la réponse

✅ Réponse : b — try/except pour gérer les exceptions.

❓ Q5. Quel module est utilisé pour le logging ?

  1. logs
  2. logging
  3. logger
💡 Voir la réponse

✅ Réponse : b — import logging

❓ Q6. Comment créer un environnement virtuel avec venv ?

  1. python -m venv mon_env
  2. pip create venv mon_env
  3. python --venv mon_env
💡 Voir la réponse

✅ Réponse : a — python -m venv nom_environnement

❓ Q7. Quel fichier liste les dépendances d’un projet Python ?

  1. packages.txt
  2. requirements.txt
  3. dependencies.json
💡 Voir la réponse

✅ Réponse : b — requirements.txt avec pip freeze

📚 Ressources pour aller plus loin

📖 Documentation officielle

🎓 Cours et tutoriels

🛠️ Outils recommandés

  • PyPI — Repository de packages Python
  • Ruff — Linter ultra-rapide
  • Black — Formateur de code
  • mypy — Vérification des types

📊 Packages Data Engineering essentiels

Package Usage
pandas Manipulation de données tabulaires
numpy Calcul numérique
requests Appels API HTTP
sqlalchemy ORM et connexions bases de données
pydantic Validation de données
pytest Tests unitaires
click CLI (Command Line Interface)

➡️ Prochaine étape

Maintenant que tu maîtrises les bases de Python, passons au traitement de données !

👉 Module suivant : 05_python_data_processing_for_data_engineers.ipynb — Pandas, Matplotlib, Seaborn et ETL

🎉 Félicitations ! Tu as terminé le module Python Basics pour Data Engineers.

Retour au sommet