J'ai essayé de résumer brièvement SQLAlchemy (il y a aussi des CONSEILS)

Cible

Résumé de SQLAlchemy, la boîte à outils SQL pour python Les questions, suggestions et demandes de modification sont les bienvenues. Je vous remercie.

Qu'est-ce que SQLAlchemy? ??

• SQLAlchemy - The Database Toolkit for Python

Outil ORM (Object Relational Mapper) utilisé lors de l'utilisation de DB en Python

Préparation environnementale

Créer un environnement avec Docker

Facile à construire avec Docker

docker-compose.yml

version: "3"

services:
  db:
    image: postgres:11.7
    container_name: sqlalchemy-db
    ports:
      - 5432:5432
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=sqlalchemy
  app:
    image: python:3.8.2-slim
    container_name: sqlalchemy-app
    volumes:
      - ./src:/usr/src
    environment:
      - PYTHONPATH=${PYTHONPATH}:/usr/src
    tty: true

Installation du package

Installez ** SQLAlchemy ** et ** psycopg2-binary **

Installer des packages python

pip install SQLAlchemy psycopg2-binary

root@4e7436ca992b:/usr/src# pip install SQLAlchemy psycopg2-binary
Collecting SQLAlchemy
  Downloading SQLAlchemy-1.3.16-cp38-cp38-manylinux2010_x86_64.whl (1.2 MB)
     |████████████████████████████████| 1.2 MB 8.3 MB/s
Collecting psycopg2-binary
  Downloading psycopg2_binary-2.8.5-cp38-cp38-manylinux1_x86_64.whl (3.0 MB)
     |████████████████████████████████| 3.0 MB 60.6 MB/s
Installing collected packages: SQLAlchemy, psycopg2-binary
Successfully installed SQLAlchemy-1.3.16 psycopg2-binary-2.8.5

Préparation de SQLAlchemy

Préparez SQLAlchemy à partir d'ici Si vous complétez facilement l'image, ce sera comme suit

Engine est un pont qui relie Python et DB (PostgresSQL) La session est comme une boîte qui contient des instructions (requête) à exécuter par DB à l'aide d'Engine. Models est une classe à associer aux informations de table définies dans DB De plus, chaque modèle hérite de la classe définie dans ** declareeative_base **.

Création d'un moteur et d'une session

Utilisez ** create_engine () ** pour créer le moteur Utilisez ** sessionmaker () ** pour créer une session

setting.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

#Créer un moteur
Engine = create_engine(
    "postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy",
    encoding="utf-8",
    echo=False
)

#Créer une session
session = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=Engine
)

BaseModel = declarative_base()

Créer un modèle

Définir UserModel Définissez uniquement les colonnes id et name ici

model.py

from sqlalchemy import Column, String, Integer
from setting import BaseModel


class UserModel(BaseModel):
    """
    UserModel
    """
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False)

    def __init__(self, name, created_at=None, updated_at=None):
        self.name = name

Utilisation de SQLAlchemy

Créer et supprimer des tables à l'aide de BaseModel

Si BaseModel est hérité par tous les modèles, les tables peuvent être créées et supprimées à l'aide des ** métadonnées ** de BaseModel. Ce que vous écrivez souvent dans SetUp et TearDown, comme lors de l'écriture de tests unitaires

Créer et supprimer des tables

from setting import Engine
from model import BaseModel

#Créer une table
BaseModel.metadata.create_all(bind=Engine)

#Supprimer la table
BaseModel.metadata.drop_all(Engine)

Insérer un enregistrement (instruction INSERT)

L'instruction INSERT utilise ** session.add () **

• Modifications terminées: ** session.commit () ** --Annuler les modifications: ** session.rollback () **
• Fin d'utilisation de la session: ** session.close () **

** SQLAlchemyError ** intercepte toutes les erreurs qui se produisent dans SQLAlchemy

Insérer un enregistrement (instruction INSERT)

from setting import Session, Engine
from model import UserModel, BaseModel
from sqlalchemy.exc import SQLAlchemyError

def model_insert():

    try:
        session = Session()

        user = UserModel(name='Maso')

        session.add(user)

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()

if __name__ == "__main__":
    #Créer une table
    BaseModel.metadata.create_all(bind=Engine)

    model_insert()

    #Supprimer la table
    BaseModel.metadata.drop_all(Engine)

Obtenir un enregistrement (instruction SELECT)

Obtenir des enregistrements en utilisant ** query () ** dans la session

• Obtenez un seul enregistrement: utilisez ** d'abord () **
• Obtenez tous les enregistrements: utilisez ** all () **
• Spécification de la condition: utilisez ** filter () **

Il y a beaucoup d'autres choses (comme la jonction de tables), il peut donc être judicieux de rechercher dans le didacticiel ci-dessous. Object Relational Tutorial — SQLAlchemy 1.3 Documentation

Obtenir un enregistrement (instruction SELECT)

def model_select():

    try:
        session = Session()

        #Obtenez un seul enregistrement
        user = session.query(UserModel).first()
        #Obtenez tous les enregistrements
        user = session.query(UserModel).all()
        #Obtention en spécifiant des conditions (clause WHERE)
        user = session.query(UserModel).\
            filter(UserModel.name == 'Maso').\
            all()

    except SQLAlchemyError:
        pass
    finally:
        session.close()

Enregistrer la mise à jour (instruction UPDATE)

Mise à jour en réécrivant les propriétés de l'enregistrement acquis en session et en validant

Enregistrer la mise à jour (instruction UPDATE)

def model_update():

    try:
        session = Session()

        #Obtention en spécifiant des conditions (clause WHERE)
        user = session.query(UserModel).\
            filter(UserModel.name == 'Maso').\
            first()

        user.name = 'Tarao'

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()

Supprimer l'enregistrement (instruction DELETE)

Il est supprimé lorsque ** delete () ** est exécuté sur l'enregistrement acquis par session.

Enregistrer la mise à jour (instruction UPDATE)

def model_delete():

    try:
        session = Session()

        session.query(UserModel).\
            filter(UserModel.name == 'Tarao').\
            .delete()

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()

Résumé du code source

main.py

from setting import Session, Engine
from model import UserModel, BaseModel
from sqlalchemy.exc import SQLAlchemyError


def model_insert():

    try:
        session = Session()

        user = UserModel(name='Maso')

        session.add(user)

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()


def model_select():

    try:
        session = Session()

        #Obtenez un seul enregistrement
        user = session.query(UserModel).first()
        #Obtenez tous les enregistrements
        user = session.query(UserModel).all()
        #Obtention en spécifiant des conditions (clause WHERE)
        user = session.query(UserModel).\
            filter(UserModel.name == 'Maso').\
            all()

    except SQLAlchemyError:
        pass
    finally:
        session.close()


def model_update():

    try:
        session = Session()

        #Obtention en spécifiant des conditions (clause WHERE)
        user = session.query(UserModel).\
            filter(UserModel.name == 'Maso').\
            first()

        user.name = 'Tarao'

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()


def model_delete():

    try:
        session = Session()

        #Obtention en spécifiant des conditions (clause WHERE)
        session.query(UserModel).\
            filter(UserModel.name == 'Tarao').\
            .delete()

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()


if __name__ == "__main__":
    #Créer une table
    BaseModel.metadata.create_all(bind=Engine)

    model_insert()

    model_select()

    model_update()

    model_delete()

    #Supprimer la table
    BaseModel.metadata.drop_all(Engine)

Tips

Je voudrais avoir une colonne commune. ??

Par exemple, la date de création (created_at) et la date de mise à jour (updated_at) Il est difficile d'écrire dans la définition du modèle à chaque fois, il est donc pratique d'utiliser ** @ declare_attr ** Passer une classe avec des colonnes créées à l'aide de @declared_attr lors de la création d'un BaseModel créera une colonne commune

Créer une colonne commune

class Base(object):
    @declared_attr
    def created_at(cls):
        return Column(DateTime, default=datetime.now, nullable=False)

    @declared_attr
    def updated_at(cls):
        return Column(DateTime, default=datetime.now, nullable=False)

BaseModel = declarative_base(cls=Base)

Est-il possible de créer un DB? ??

Facile à créer et à supprimer à l'aide de SQLAlchemy-Utils - documentation SQLAlchemy-Utils 0.33.9 Image à utiliser au début du test ou lors du lancement de l'application

Vérification et création de l'existence d'une base de données

from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database, drop_database
engine = create_engine("postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy")
    if not database_exists(engine.url):
        #Création de DB
        create_database(engine.url)
    #Supprimer la base de données
    drop_database(engine.url)

Comment créer un schéma? ??

Si vous utilisez postgres etc. et que vous souhaitez créer un schéma, utilisez CreateSchema et DropSchema

Créer / supprimer un schéma

from sqlalchemy.schema import CreateSchema, DropSchema
from sqlalchemy import create_engine

engine = create_engine("postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy")
schema_name = 'test'
#Vérification de l'existence du schéma
if not engine.dialect.has_schema(Engine, schema_name):
    #Créer un schéma
    engine.execute(CreateSchema(schema_name))

#Supprimer le schéma
Engine.execute(DropSchema(schema_name))