Accès aux Données (Supabase)

Ce fichier documente comment le backend StageConnect accède à la base de données via le client Supabase.

Les Deux Clients Supabase

Le projet utilise deux clients distincts selon le contexte :

Client	Fonction	Clé utilisée	RLS respecté
supabase	Opérations utilisateur normales	SUPABASE_ANON_KEY	✅ Oui
supabase_admin	Opérations privilégiées	SUPABASE_SERVICE_ROLE_KEY	❌ Non

Configuration

Fichier : app/db/supabase.py

from supabase import create_client, Client
from app.core.config import settings

# Client normal — respecte les politiques RLS
supabase: Client = create_client(SUPABASE_URL, SUPABASE_ANON_KEY)

# Client admin — bypass les politiques RLS
supabase_admin: Client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)

Quand utiliser quel client

Utiliser supabase (anon) :

• Lecture de données utilisateur
• Insert/Update avec contraintes RLS
• Opérations initiées par un utilisateur authentifié

Utiliser supabase_admin :

• Création de profils utilisateur lors de l'inscription
• Opérations cross-tenant (admin qui modifie les données d'autres organisations)
• Insertion initiale dans des tables avec RLS strictes

Pattern d'Accès aux Données

Tous les exemples ci-dessous sont tirés du code existant.

SELECT avec jointures

app/services/student/student_service.py

query = supabase.table("students").select(
    "*, "
    "academic_programs(program_name, degree_level), "
    "departments(name, field_of_study), "
    "users(email, phone), "
    "student_themes(title), "
    "internship_conventions!internship_conventions_student_id_fkey("
        "status, "
        "end_date, "
        "companies(company_name), "
        "academic_staff(first_name, last_name)"
    ")",
    count="exact"
).eq("university_id", university_id).eq("is_active", True)

Points clés :

• Syntaxe table!foreign_key_fkey(columns) pour les jointures
• .count="exact" pour récupérer le nombre total
• Chaînage de .eq() pour les filtres

SELECT avec filtres multiples

app/services/student/student_pair_service.py

# Filtre avec .eq() chainés
students = supabase.table("students").select(
    "id, first_name, last_name, matricule_student"
).eq("university_id", university_id).eq(
    "school_faculty_id", faculty_id
).execute()

# Filtre avec .or_()
pair = supabase.table("student_pairs").select("*").or_(
    f"student_1_id.eq.{sid},student_2_id.eq.{sid}"
).eq("status", "confirmed").execute()

INSERT

app/services/student/student_pair_service.py

pair = supabase.table("student_pairs").insert({
    "student_1_id": sid,
    "student_2_id": partner_student_id,
    "student_theme_id": tid,
    "initiated_by_student_id": sid,
    "status": "confirmed",
    "confirmation_date": datetime.utcnow().isoformat()
}).execute()

UPDATE

app/services/student/student_pair_service.py

res = supabase.table("student_pairs").update({
    "status": "dissolved",
    "dissolution_date": datetime.utcnow().isoformat()
}).eq("id", pair_id).or_(
    f"student_1_id.eq.{user_id},student_2_id.eq.{user_id}"
).execute()

DELETE

app/api/v1/routers/students/documents.py

result = supabase.table("student_documents").delete().eq(
    "id", doc_id
).eq("student_id", student_id).execute()

Pagination avec range

app/api/v1/routers/students/documents.py

offset = (page - 1) * limit
response = supabase.table("student_documents").select(
    "*", count="exact"
).eq("student_id", student_id).range(offset, offset + limit - 1).execute()

Méthodes de Filtre Courantes

Méthode	Usage	Exemple
.eq()	Égalité exacte	.eq("status", "pending")
.neq()	Différent de	.neq("id", excluded_id)
.gt() / .gte()	Supérieur (strict / égal)	.gt("created_at", threshold)
.lt() / .lte()	Inférieur (strict / égal)	.lte("age", 18)
.like() / .ilike()	Like SQL (case-sensitive / insensitive)	.ilike("name", "%test%")
.or_()	Condition OR	.or_("status.eq.pending,status.eq.review")
.in_()	Dans une liste	.in_("id", [1, 2, 3])
.is_()	Null check	.is_("deleted_at", None)
.range()	Pagination	.range(0, 9)

Gestion des Erreurs

Vérifier les données retournées

response = supabase.table("students").select("*").execute()

# Toujours vérifier si des données existent
if response.data:
    students = response.data

# Pour les comptages
total = response.count  # Si count="exact" demandé

Try/Except

try:
    response = supabase.table("students").select("*").execute()
    students = response.data or []
except Exception as e:
    print(f"Erreur fetching students: {e}")
    students = []

Erreurs Supabase

Le SDK Supabase lève une exception si la requête échoue. Toujours wrapped les appels dans un try/except.

Règles Absolues

1. Jamais d'input utilisateur dans les noms de fonctions ou de tables

Interdit — construction dynamique avec input utilisateur :

supabase.rpc(user_input)  # jamais un nom de fonction venant de l'user
supabase.table(user_input)  # jamais un nom de table venant de l'user

Correct — noms fixes codés en dur :

supabase.rpc("get_next_recommendation_number")  # nom fixe, sécurisé
supabase.table("students").select("*").eq("email", email)  # paramètre bindé

remarque

Les 3 appels supabase.rpc() du projet utilisent des noms de fonctions fixes — aucune variable utilisateur. C'est la bonne pratique à maintenir.

2. Toujours utiliser le bon client

Créer un profil utilisateur → supabase_admin :

app/services/auth/creators/student_creator.py

# Création de profil — nécessite admin pour bypass RLS
res = supabase_admin.table("students").insert(record).execute()

Lire ses propres données → supabase :

# Lecture standard — respecte RLS
profile = supabase.table("students").select("*").eq("user_id", user_id).execute()

3. Documenter l'utilisation de supabase_admin

Chaque utilisation de supabase_admin doit être justifiée dans un commentaire :

# Admin requis : la table students a une politique RLS stricte
# qui empêche l'auto-insertion lors de l'inscription
res = supabase_admin.table("students").insert(record).execute()

Ajouter un Nouvel Accès à une Table

Étape 1 : Créer le service

Créer un nouveau fichier dans app/services/{domain}/ :

app/services/company/
├── __init__.py
├── offer_manager.py      # Offres de stage
└── new_service.py        # Nouveau service

Étape 2 : Importer le bon client

from app.db.supabase import get_supabase_client, get_supabase_admin_client

class NewService:
    def __init__(self):
        self.supabase = get_supabase_client()
        self.supabase_admin = get_supabase_admin_client()

Étape 3 : Implémenter les méthodes

def get_items(self, user_id: str):
    """Lecture — utilise le client normal (respecte RLS)"""
    return self.supabase.table("items").select(
        "*"
    ).eq("owner_id", user_id).execute()

def create_item(self, data: dict):
    """Création — utilise admin car RLS bloque l'insertion"""
    return self.supabase_admin.table("items").insert(data).execute()

Étape 4 : Tester les deux scénarios

1. Avec un utilisateur normal : vérifier que RLS est respecté
2. Avec un admin : vérifier que le bypass fonctionne

Étape 5 : Utiliser dans un router

app/api/v1/routers/companies/new.py

from app.services.company.new_service import NewService

router = APIRouter()
service = NewService()

@router.get("/items")
def get_items(current_user = Depends(get_current_user)):
    return service.get_items(current_user.id)

Fichiers de Référence

Fichier	Usage
app/db/supabase.py	Configuration des clients
app/services/student/student_service.py	SELECT avec jointures complexes
app/services/student/student_pair_service.py	INSERT, UPDATE, DELETE
app/services/auth/creators/student_creator.py	Utilisation de supabase_admin