Zusammenfassung von SQLAlchemy, dem SQL-Toolkit für Python Fragen, Vorschläge und Bearbeitungswünsche sind willkommen. Vielen Dank.
ORM-Tool (Object Relational Mapper), das bei der Arbeit mit DB in Python verwendet wird
Einfach mit Docker zu bauen
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
Installieren Sie ** SQLAlchemy ** und ** psycopg2-binary **
Python-Pakete installieren
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
Bereiten Sie von hier aus SQLAlchemy vor Wenn Sie das Bild leicht ergänzen, sieht es wie folgt aus
Engine ist eine Brücke, die Python und DB verbindet (PostgresSQL) Die Sitzung ist wie eine Box, in der Anweisungen (Abfragen) gepackt sind, die von der DB mithilfe der Engine ausgeführt werden sollen. Models ist eine Klasse zum Verknüpfen mit in DB definierten Tabelleninformationen Darüber hinaus erbt jedes Modell die in [** decleative_base **] definierte Klasse (https://docs.sqlalchemy.org/en/13/orm/extensions/declarative/api.html).
Verwenden Sie ** create_engine () **, um die Engine zu erstellen Verwenden Sie ** sessionmaker () **, um eine Sitzung zu erstellen
setting.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
#Erstellen einer Engine
Engine = create_engine(
"postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy",
encoding="utf-8",
echo=False
)
#Sitzung erstellen
session = sessionmaker(
autocommit=False,
autoflush=False,
bind=Engine
)
BaseModel = declarative_base()
Definieren Sie UserModel Definieren Sie hier nur die Spalten id und name
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
Wenn BaseModel von allen Modellen geerbt wird, können Tabellen mithilfe der ** Metadaten ** von BaseModel erstellt und gelöscht werden. Dinge, die Sie häufig in SetUp und TearDown schreiben, z. B. beim Schreiben von Komponententests
Tabellen erstellen und löschen
from setting import Engine
from model import BaseModel
#Eine Tabelle erstellen
BaseModel.metadata.create_all(bind=Engine)
#Tabelle löschen
BaseModel.metadata.drop_all(Engine)
Die INSERT-Anweisung verwendet ** session.add () **
** SQLAlchemyError ** fängt alle Fehler ab, die in SQLAlchemy auftreten
Datensatz einfügen (INSERT-Anweisung)
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__":
#Eine Tabelle erstellen
BaseModel.metadata.create_all(bind=Engine)
model_insert()
#Tabelle löschen
BaseModel.metadata.drop_all(Engine)
Holen Sie sich Datensätze mit ** query () ** in der Sitzung
Es gibt viele andere Dinge (wie das Verbinden von Tabellen), daher ist es möglicherweise eine gute Idee, im folgenden Tutorial zu suchen. Object Relational Tutorial — SQLAlchemy 1.3 Documentation
Datensatz abrufen (SELECT-Anweisung)
def model_select():
try:
session = Session()
#Holen Sie sich nur einen Datensatz
user = session.query(UserModel).first()
#Holen Sie sich alle Aufzeichnungen
user = session.query(UserModel).all()
#Erhalten Sie durch Angabe von Bedingungen (WHERE-Klausel)
user = session.query(UserModel).\
filter(UserModel.name == 'Maso').\
all()
except SQLAlchemyError:
pass
finally:
session.close()
Aktualisiert durch Umschreiben der Eigenschaften des in der Sitzung erfassten Datensatzes und Festschreiben
Datensatzaktualisierung (UPDATE-Anweisung)
def model_update():
try:
session = Session()
#Erhalten Sie durch Angabe von Bedingungen (WHERE-Klausel)
user = session.query(UserModel).\
filter(UserModel.name == 'Maso').\
first()
user.name = 'Tarao'
session.commit()
except SQLAlchemyError:
session.rollback()
finally:
session.close()
Es wird gelöscht, wenn ** delete () ** für den von der Sitzung erfassten Datensatz ausgeführt wird.
Datensatzaktualisierung (UPDATE-Anweisung)
def model_delete():
try:
session = Session()
session.query(UserModel).\
filter(UserModel.name == 'Tarao').\
.delete()
session.commit()
except SQLAlchemyError:
session.rollback()
finally:
session.close()
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()
#Holen Sie sich nur einen Datensatz
user = session.query(UserModel).first()
#Holen Sie sich alle Aufzeichnungen
user = session.query(UserModel).all()
#Erhalten Sie durch Angabe von Bedingungen (WHERE-Klausel)
user = session.query(UserModel).\
filter(UserModel.name == 'Maso').\
all()
except SQLAlchemyError:
pass
finally:
session.close()
def model_update():
try:
session = Session()
#Erhalten Sie durch Angabe von Bedingungen (WHERE-Klausel)
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()
#Erhalten Sie durch Angabe von Bedingungen (WHERE-Klausel)
session.query(UserModel).\
filter(UserModel.name == 'Tarao').\
.delete()
session.commit()
except SQLAlchemyError:
session.rollback()
finally:
session.close()
if __name__ == "__main__":
#Eine Tabelle erstellen
BaseModel.metadata.create_all(bind=Engine)
model_insert()
model_select()
model_update()
model_delete()
#Tabelle löschen
BaseModel.metadata.drop_all(Engine)
Tips
Zum Beispiel das Erstellungsdatum (created_at) und das Aktualisierungsdatum (update_at) Es ist schwierig, jedes Mal in die Modelldefinition zu schreiben, daher ist es bequem, ** @ declare_attr ** zu verwenden Wenn Sie beim Erstellen eines BaseModel eine Klasse mit Spalten übergeben, die mit @declared_attr erstellt wurden, wird eine gemeinsame Spalte erstellt
Erstellen Sie eine gemeinsame Spalte
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)
Einfach zu erstellen und zu löschen mit Dokumentation zu SQLAlchemy-Utils - SQLAlchemy-Utils 0.33.9 Bild, das zu Beginn des Tests oder beim Starten der Anwendung verwendet werden soll
DB Existenzprüfung und Erstellung
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):
#DB-Erstellung
create_database(engine.url)
#DB löschen
drop_database(engine.url)
Wenn Sie Postgres usw. verwenden und ein Schema erstellen möchten, verwenden Sie CreateSchema und DropSchema
Schema erstellen / löschen
from sqlalchemy.schema import CreateSchema, DropSchema
from sqlalchemy import create_engine
engine = create_engine("postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy")
schema_name = 'test'
#Überprüfung der Schemaexistenz
if not engine.dialect.has_schema(Engine, schema_name):
#Erstellen eines Schemas
engine.execute(CreateSchema(schema_name))
#Schema löschen
Engine.execute(DropSchema(schema_name))
Recommended Posts