I tried to summarize SQLAlchemy briefly (There is also TIPS)

Target

Summary of SQLAlchemy, python's SQL toolkit Questions, suggestions and editing requests are welcome. Thank you.

What is SQLAlchemy? ??

ORM (Object Relational Mapper) tool used when manipulating DB in Python

Environmental preparation

Build an environment with Docker

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

Package installation

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

Preparing SQLAlchemy

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 **.

Creating Engine and Session

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()

Creating a Model

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

Using SQLAlchemy

Create and delete tables using BaseModel

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 record (INSERT statement)

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 record (SELECT statement)

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()

Record update (UPDATE statement)

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()

Delete record (DELETE statement)

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()

Source code summary

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

I would like to have a common column. ??

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)

Is it possible to create a DB? ??

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)

How to create a schema? ??

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

I tried to summarize SQLAlchemy briefly (There is also TIPS)
I tried to summarize SparseMatrix
I tried to summarize Python exception handling
I tried to summarize the umask command
Python3 standard input I tried to summarize
I tried to summarize Ansible modules-Linux edition
LeetCode I tried to summarize the simple ones
I tried to debug.
I tried to paste
I tried to summarize how to use matplotlib of python
I tried to summarize the basic form of GPLVM
I tried to summarize four neural network optimization methods
I tried to summarize how to use pandas in python
I want to say that there is data preprocessing ~
I tried to summarize the string operations of Python
[First COTOHA API] I tried to summarize the old story
I tried to learn PredNet
I tried to organize SVM.
I tried to implement PCANet
I tried to summarize the code often used in Pandas
I tried to reintroduce Linux
I tried to summarize the commands often used in business
I tried to touch jupyter
I tried to implement StarGAN (1)
I tried to summarize how to use the EPEL repository again
I tried to create an article in Wiki.js with SQLAlchemy
[Linux] I tried to summarize the command of resource confirmation system
I tried using the COTOHA API (there is code on GitHub)
I tried to summarize what was output with Qiita with Word cloud
I tried to find out if ReDoS is possible with Python
I tried to summarize the commands used by beginner engineers today
I tried to summarize everyone's remarks on slack with wordcloud (Python)
I tried to summarize the frequently used implementation method of pytest-mock
I tried to implement Deep VQE
I tried to create Quip API
I tried to touch Python (installation)
I tried to implement adversarial validation
I tried to explain Pytorch dataset
I tried Watson Speech to Text
I tried to touch Tesla's API
I tried to implement hierarchical clustering
I tried to organize about MCMC.
I tried to implement Realness GAN
I tried to move the ball
I tried to estimate the interval.
I tried to summarize until I quit the bank and became an engineer
I tried to implement deep learning that is not deep with only NumPy
I tried to analyze emotions whether Hinatazaka46 is really a "happy aura"
I tried to summarize Cpaw Level1 & Level2 Write Up in an easy-to-understand manner
I tried to summarize various sentences using the automatic summarization API "summpy"
I tried to summarize the logical way of thinking about object orientation.
I tried to summarize the Linux commands used by beginner engineers today-Part 1-
I tried to find out what I can do because slicing is convenient
I tried to implement PLSA in Python
I tried using Azure Speech to Text.
I tried to implement Autoencoder with TensorFlow
I tried to implement permutation in Python
I tried to create a linebot (preparation)
I tried to visualize AutoEncoder with TensorFlow
I tried to recognize the wake word
I tried to get started with Hy