Summary of SQLAlchemy, python's SQL toolkit Questions, suggestions and editing requests are welcome. Thank you.
ORM (Object Relational Mapper) tool used when manipulating DB in Python
Easy to build with 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
Install ** SQLAlchemy ** and ** psycopg2-binary **
Install python packages
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
Prepare SQLAlchemy from here If you supplement the image easily, it will be as follows
Engine is a bridge connecting Python and DB (PostgresSQL) Session is like a box that packs instructions (Query) to be executed by DB using Engine. Models is a class for associating with table information defined in DB Also, each Model inherits the class defined in ** declareative_base **.
Use ** create_engine () ** to create the Engine Use ** sessionmaker () ** to create a Session
setting.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
#Creating an Engine
Engine = create_engine(
"postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy",
encoding="utf-8",
echo=False
)
#Creating a Session
session = sessionmaker(
autocommit=False,
autoflush=False,
bind=Engine
)
BaseModel = declarative_base()
Define UserModel Only define the id and name columns here
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
If BaseModel is inherited by all models, you can create and delete tables using ** metadata ** of BaseModel. Things you often write in SetUp and TearDown, such as when writing unit tests
Create and delete tables
from setting import Engine
from model import BaseModel
#Creating a table
BaseModel.metadata.create_all(bind=Engine)
#Delete table
BaseModel.metadata.drop_all(Engine)
INSERT statement uses ** session.add () **
--Completed changes: ** session.commit () ** --Undo changes: ** session.rollback () ** --End of session: ** session.close () **
** SQLAlchemyError ** catches all errors that occur in SQLAlchemy
Insert record (INSERT statement)
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__":
#Creating a table
BaseModel.metadata.create_all(bind=Engine)
model_insert()
#Delete table
BaseModel.metadata.drop_all(Engine)
Get records using ** query () ** in session
--Get only one record: Use ** first () ** --Get all records: Use ** all () ** --Condition specification: Use ** filter () **
There are many other things (such as joining tables), so it might be a good idea to search from the tutorial below. Object Relational Tutorial — SQLAlchemy 1.3 Documentation
Get record (SELECT statement)
def model_select():
try:
session = Session()
#Get only one record
user = session.query(UserModel).first()
#Get all records
user = session.query(UserModel).all()
#Get by specifying the condition (WHERE clause)
user = session.query(UserModel).\
filter(UserModel.name == 'Maso').\
all()
except SQLAlchemyError:
pass
finally:
session.close()
It is updated when you rewrite the property of the record acquired in session and commit
Record update (UPDATE statement)
def model_update():
try:
session = Session()
#Get by specifying the condition (WHERE clause)
user = session.query(UserModel).\
filter(UserModel.name == 'Maso').\
first()
user.name = 'Tarao'
session.commit()
except SQLAlchemyError:
session.rollback()
finally:
session.close()
It is deleted when ** delete () ** is executed for the record acquired by session.
Record update (UPDATE statement)
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()
#Get only one record
user = session.query(UserModel).first()
#Get all records
user = session.query(UserModel).all()
#Get by specifying the condition (WHERE clause)
user = session.query(UserModel).\
filter(UserModel.name == 'Maso').\
all()
except SQLAlchemyError:
pass
finally:
session.close()
def model_update():
try:
session = Session()
#Get by specifying the condition (WHERE clause)
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()
#Get by specifying the condition (WHERE clause)
session.query(UserModel).\
filter(UserModel.name == 'Tarao').\
.delete()
session.commit()
except SQLAlchemyError:
session.rollback()
finally:
session.close()
if __name__ == "__main__":
#Creating a table
BaseModel.metadata.create_all(bind=Engine)
model_insert()
model_select()
model_update()
model_delete()
#Delete table
BaseModel.metadata.drop_all(Engine)
Tips
For example, the creation date (created_at) and the update date (updated_at) It's hard to write in the model definition every time, so it's convenient to use ** @ declared_attr ** Passing a class with columns created using @declared_attr when creating a BaseModel will create a common column
Create a common column
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)
Easy to create and delete using SQLAlchemy-Utils — SQLAlchemy-Utils 0.33.9 documentation Image to use at the beginning of the test or when launching the application
DB existence check and creation
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 creation
create_database(engine.url)
#Delete DB
drop_database(engine.url)
If you are using postgres etc. and want to create a schema, use CreateSchema and DropSchema
Create / delete schema
from sqlalchemy.schema import CreateSchema, DropSchema
from sqlalchemy import create_engine
engine = create_engine("postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy")
schema_name = 'test'
#Schema existence check
if not engine.dialect.has_schema(Engine, schema_name):
#Creating a schema
engine.execute(CreateSchema(schema_name))
#Schema deletion
Engine.execute(DropSchema(schema_name))
Recommended Posts