Aller au contenu

Transaction par

Les transactions de base de données sont prises en charge grâce à l'encodage/base de données utilisée pour émettre des requêtes asynchrones.

Usage basique

Pour utiliser les transactions, utilisez database.transaction comme gestionnaire de contexte asynchrone:

1
2
3
4
5
async avec database.transaction():
    #Tout appelé ici sera une transaction
    await Model1().save()
    await Model2().save()
    ...

!!!Notez qu 'il doit s'agir de la même base de données que celle utilisée dans l'objet ormar_config du modèle.

Pour éviter de passer des instances de base de données dans votre code, vous pouvez extraire des instances de chaque modèle. les bases de données fournies lors de la déclaration ormar.Model sont disponibles via la base de données ormar_config.database et sont accessibles depuis les classes et les instances.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
Importer des databases
Import sqlalchemy
Importer à Ormar


base_ormar_config = OrmarConfig(
    metadata=sqlalchemy.MetaData(),
    database = databases.Database("sqlite://"),
)


Auteur de classe (modèle ormar):
    ormar_config = base_ormar_config.copy()

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=255)

# database est accessible depuis la classe
database = Author.ormar_config.database

#Aussi bien que de l'instance
Author = Author(nom ="Stephen King")
database = author.ormar_config.database

Vous pouvez également utiliser.transaction() comme décorateur de fonction pour toute fonction asynchrone:

1
2
3
@database.transaction()
async def create_users(demande):
    ...

Les blocs de transactions sont gérés comme des états locaux de tâche. les transactions imbriquées sont entièrement prises en charge et sont implémentées à l'aide de points de sauvegarde de base de données.

Soumission manuelle/rollback

Pour les API de trading de niveau inférieur, vous pouvez le déclencher manuellement.

1
2
3
4
5
6
7
8
transaction = await database.transaction()
Essayez:
    await transaction.start()
    ...
Excepté:
    await transaction.rollback()
Else:
    await transaction.commit()

Tester le

Les transactions sont également utiles pendant les tests lorsque vous pouvez appliquer une restauration forcée et que vous n'avez pas à nettoyer les données après chaque test.

1
2
3
4
5
6
@pytest.mark.asyncio
async def sample_test:
    async avec base de donnée:
        async avec database.transaction(force_rollback=True):
            #Votre code de test ici
            ...