sqlite-persist 0.3.0

Classe de base ORM légère pour SQLite

sqlite-persist

Classe de base ORM légère pour SQLite, sans dépendances externes.

Installation

# depuis un dépôt git
uv add git+https://codeberg.org/styfore/sqlite-persist.git
# ou en local
uv add sqlite-persist

Mise en place

from sqlite_persist import Persistence # AsyncPersistence pour la version async, et adapater avec des await, async, etc
import sqlite3

con = sqlite3.connect("ma_base.db")

class User(Persistence):
    def __init__(self, id: int | None, username: str, email: str, age: int):
        self.id = id
        self.username = username
        self.email = email
        self.age = age

Lecture

# Par PK — lève KeyError si introuvable
user = User.get(1, con=con)

# Par PK — retourne None si introuvable
user = User.find(1, con=con)

# Sélection avec filtres
users = User.select(con=con)                                      # SELECT *
users = User.select({"username": "alice"}, con=con)               # opérateur = par défaut
users = User.select({"username LIKE": "ali%"}, con=con)           # opérateur custom
users = User.select({"age >=": 18, "age <": 30}, con=con)         # même colonne, deux opérateurs
users = User.select({"username": "alice"}, order_by="age", con=con)
users = User.select({"username": "alice"}, order_by=["age", "email DESC"], con=con)

# Requête libre (JOIN, sous-requêtes, etc.)
users = User.select(
    "username LIKE :q OR email LIKE :q",
    params={"q": "%alice%"},
    con=con,
)

# Requête SQL complète
users = User.from_query("""
    SELECT u.*, COUNT(o.id) as nb_orders
    FROM user u
    LEFT JOIN orders o ON o.user_id = u.id
    GROUP BY u.id
    HAVING nb_orders > :min
""", params={"min": 2}, con=con)
# Les colonnes extra (nb_orders) sont accessibles dans __dict__,
# attention un update après lèvera une erreur — nommer la colonne supplémentaire
# en commençant par _ pour l'exclure de _row_data

# Comptage et existence
count = User.count(con=con)                         # SELECT COUNT(*)
count = User.count({"age >=": 18}, con=con)         # avec filtre dict
count = User.count("age >= :a", {"a": 18}, con=con) # avec requête libre

exists = User.exists({"username": "alice"}, con=con)   # True / False
exists = User.exists("age >= :a", {"a": 18}, con=con)

Écriture

# Insert — met à jour self.id avec la valeur générée par SQLite
user = User(None, "alice", "[e-mail]", 30)
user.insert(con=con)
print(user.id)  # id généré

# Update strict — la ligne doit exister
user.age = 31
user.update(con=con)

# Upsert — INSERT OR REPLACE, gère les deux cas
user.upsert(con=con)

# Delete
user.delete(con=con)

Opérations batch

users = [
    User(None, "alice", "[e-mail]", 30),
    User(None, "bob",   "[e-mail]",   25),
]

User.insert_all(users, con=con)   # ids mis à jour sur chaque item, atomique
User.update_all(users, con=con)
User.upsert_all(users, con=con)   # ids récupérés si absents
User.delete_all(users, con=con)

# Sans instances
User.update_where(values={"age": 0}, where={"username": "bob"}, con=con)
User.delete_where({"username": "bob"}, con=con)

Transactions

Grouper plusieurs opérations dans une transaction atomique — commit automatique à la sortie du bloc, rollback en cas d'exception :

with User.transaction(con) as tx:
    u1 = User(None, "alice", "[e-mail]", 30)
    u2 = User(None, "bob",   "[e-mail]",   25)
    u1.insert(con=tx)
    u2.insert(con=tx)
# commit ici

# rollback automatique si exception
try:
    with User.transaction(con) as tx:
        u.insert(con=tx)
        u.insert(con=tx)  # IntegrityError → rollback, rien n'est inséré
except sqlite3.IntegrityError:
    pass

Les opérations batch (insert_all, upsert_all, etc.) sont elles-mêmes atomiques — elles utilisent transaction() en interne.

Injection automatique de connexion (ex. Flask)

renseigner _database au niveau de la classe

Possibilité d’utiliser un contexte manager avec la connection créer automatiquement dans ce cas.

from sqlite_persist import Persistence
from mon_app.db import get_db

class User(AppPersistence):
    _database = "path/to/database.db"

    ...

with User.connection():
    user = User.get(1)       # con= injecté automatiquement
    user.age = 31
    user.update()

# la connection est fermée à la fin du context

Cette connexion n’est utilisée que dans ce cas là et ne dispense pas de renseigner le paramètre con= ou de surcharger _con dans les autres utilisations, car il n’y a que dans le contexte que l’ouverture et fermeture peut-être gérée automatiquement

Surcharger _con() dans une classe intermédiaire — plus besoin de passer con= partout :

from sqlite_persist import Persistence
from mon_app.db import get_db

class AppPersistence(Persistence):
    @classmethod
    def _con(cls):
        return get_db()

class User(AppPersistence):
    ...

user = User.get(1)       # con= injecté automatiquement
user.age = 31
user.update()

Nom de table personnalisé

Par défaut le nom de la table est le nom de la classe. Pour le surcharger :

class User(AppPersistence):
    _table = "app_user"

Clé primaire composite

class Conge(AppPersistence):
    _pk = ("calendrier_id", "date_conge")

conge = Conge.get(42, "2025-06-01", con=con)

Attributs transitoires

Tout attribut commençant par _ est ignoré lors des opérations SQL :

class User(AppPersistence):
    def __init__(self, id, username, email, age):
        self.id = id
        self.username = username
        self.email = email
        self.age = age
        self._cache = {}     # jamais envoyé en base
        self._dirty = False  # jamais envoyé en base

Représentation

__repr__ par défaut affiche la table, les colonnes PK puis le reste :

User[id=1 | username='alice', email='[e-mail]', age=30]