How to use SQLAlchemy / Connect with aiomysql

This is a SQLAlchemy memo that allows you to write python-like code around sql. By introducing such a library, you can create a service with few bugs by eliminating SQL syntax mistakes.

The explanation below is based on version 1.1.

Overall flow

  1. Define a table
  2. Create a query from a table instance
  3. Specify the engine (DB execution engine)
  4. Pass the query to the engine and execute

Define a table

SQLAlchemy makes it possible to handle column information programmatically by defining a table in the code. You can also shift the responsibility of creating the table from manual to code.

http://docs.sqlalchemy.org/en/rel_1_1/core/metadata.html

import sqlalchemy as sa

user = sa.Table('user', metadata,
    sa.Column('user_id', sa.Integer, primary_key=True),
    sa.Column('user_name', sa.String(16), nullable=False),
    sa.Column('email_address', sa.String(60)),
    sa.Column('password', sa.String(20), nullable=False)
)

How to specify autoincrement

As you can see in AUTO_INCREMENT Behavior, autoincrement is automatically added to the first column that meets the following conditions: Will be done.

--primary key --Integer --Not a foreign key

How to specify unsigned

Use sqlalchemy.dialects.mysql.INTEGER. dialect: It's a dialect. mysql dialect. If you want to use various mysql, it may be better to use the type under dialects.mysql.

import sqlalchemy as sa
from sqlalchemy.dialects.mysql import INTEGER
...

sa.Column('scenario_id', INTEGER(unsigned=True), nullable=False)

Others are as in MySQL Data Types

from sqlalchemy.dialects.mysql import \
        BIGINT, BINARY, BIT, BLOB, BOOLEAN, CHAR, DATE, \
        DATETIME, DECIMAL, DECIMAL, DOUBLE, ENUM, FLOAT, INTEGER, \
        LONGBLOB, LONGTEXT, MEDIUMBLOB, MEDIUMINT, MEDIUMTEXT, NCHAR, \
        NUMERIC, NVARCHAR, REAL, SET, SMALLINT, TEXT, TIME, TIMESTAMP, \
        TINYBLOB, TINYINT, TINYTEXT, VARBINARY, VARCHAR, YEAR

And so on.

How to specify default (server_default)

There are two ways to specify default and server_default.

--default: Give default value if no value is specified in python layer --Does not affect the create statement (probably) --server_default: The so-called SQL default statement. --Create a default statement in create table.

Here, we will explain server_default.

server_default

sa.Column('x', sa.Text, server_default="val")
↓
x TEXT DEFAULT 'val'

sa.Column('y', sa.DateTime, server_default=sa.text('NOW()'))
↓
y DATETIME DEFAULT NOW()

The value given as in the example above is enclosed in quotation marks. If you don't want to quote it, use text.

In the usual case of creating_datetime, the default is as follows.

import sqlalchemy as sa
from sqlalchemy.dialects.mysql import DATETIME

sa.Column('create_datetime', DATETIME(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),

How to specify on update (server_onupdate)

Like default, there are onupdate and server_onupdate. The difference is the same as default, use server_onupdate to change the create table statement.

server_onupdate

A common timestamp (update time) can be written as:

import sqlalchemy as sa
from sqlalchemy.dialects.mysql import DATETIME

sa.Column('timestamp', DATETIME(), nullable=False,
          server_default=sa.text('CURRENT_TIMESTAMP'), server_onupdate=sa.text('CURRENT_TIMESTAMP'))

How to assemble a query

table = sa.Table('user', metadata,
    sa.Column('user_id', sa.Integer, primary_key=True),
    sa.Column('user_name', sa.String(16), nullable=False),
    sa.Column('email_address', sa.String(60)),
    sa.Column('password', sa.String(20), nullable=False)
)

Suppose a variable called table is defined as above.

We're just making a query here, so it won't work unless we actually put it into the DB engine. Please catch other documents for how to put it in. (I will not write it here because it seems that it is different from the general usage because I am using aiomysql)

select

import sqlalchemy as sa

q = sa.select(['user_id', 'user_name']).where(table.c.user_id == 1234)
# or
q = table.select().where(table.c.user_id == 1234)

You can specify the condition in the list of columns you want to get in select (), followed by where. table.c. represents a column. This means that the value in the user_id column is 1234.

It seems that the column name cannot be specified in table.select ().

Read ↓ for details. http://docs.sqlalchemy.org/en/rel_1_1/core/selectable.html

Get the argument of query

By the way, if you want to use the assigned parameters in the test, you can do as follows.

print(str(q))
# SELECT user.user_id, user.user_name 
# FROM user 
# WHERE user.user_id = :user_id_1

print(q.compile().params)
# {'user_id_1': 1234}

insert

q = table.insert().values(
    # user_id is auto increment
    user_name='hal',
    #Since email is not specified, it will be null
    password='greatpassword'
)

You can create a query with values specified by insert (). values (). it's easy.

Get the argument of query

If you want to use the assigned parameters etc. in the test, you can do as follows.

print(str(q))
# INSERT INTO user (user_name, password) VALUES (:user_name, :password)

print(q.compile().params)
# {'user_name': 'hal', 'password': 'greatpassword'}

Actually run using the aiomysql engine

Here's how to use aiomysql to execute the query created above. ** It is common to build an engine among the functions of sqlalchemy **, so if you want to use it normally, read here. I think it's good.

Since we are using aiomysql with asyncio, which is a new specification that does asynchronous processing well, I will briefly touch on how to use it in that case. The official documentation is http://aiomysql.readthedocs.io/en/latest/sa.html.

It is convenient to make something wrapped in the context manager as shown below.

import asyncio
from aiomysql.sa import create_engine


class MyDB:
    async def __aenter__(self):
        loop = asyncio.get_event_loop()
        config = self._load_db_config()
        engine = await create_engine(  #Make an engine with aiomysql
            host=config['host'],  #The argument can be the one of aiomysql connect
            port=config['port'],
            user=config['user'],
            password=config['password'],
            db=config['database'],
            charset='utf8',
            autocommit=True,  #If this is set to True, it will be reflected immediately when the insert command is executed.
            loop=loop
        )
        self._connection = await engine.acquire()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self._connection.close() 

    async def execute(self, query, *multiparams, **params):
        return await self._connection.execute(query, *multiparams, **params)

User side

table = sa.Table('user', metadata,
    sa.Column('user_id', sa.Integer, primary_key=True),
    sa.Column('user_name', sa.String(16), nullable=False),
    sa.Column('email_address', sa.String(60)),
    sa.Column('password', sa.String(20), nullable=False)
)


async with MyDB() as db:
    q = table.select(['user_id', 'user_name']).where(table.c.user_id == 1234)
    row_list = db.execute(q).fetchall()
    for row in row_list:
        print(row[table.c.user_name])

Recommended Posts

How to use SQLAlchemy / Connect with aiomysql
How to update with SQLAlchemy?
How to Alter with SQLAlchemy?
How to Delete with SQLAlchemy?
Connect to multiple databases with SQLAlchemy
How to use virtualenv with PowerShell
How to use FTP with Python
How to INNER JOIN with SQLAlchemy
How to get parent id with sqlalchemy
How to use ManyToManyField with Django's Admin
How to use OpenVPN with Ubuntu 18.04.3 LTS
How to use Cmder with PyCharm (Windows)
How to use Ass / Alembic with HtoA
How to use Japanese with NLTK plot
How to use jupyter notebook with ABCI
How to use CUT command (with sample)
How to use JDBC driver with Redash
How to use xml.etree.ElementTree
How to use Python-shell
How to use tf.data
How to use virtualenv
How to use Seaboan
How to use image-match
How to use shogun
How to use Virtualenv
How to use numpy.vectorize
How to use pytest_report_header
How to use partial
How to use Bio.Phylo
How to use SymPy
How to use x-means
How to use WikiExtractor.py
How to use virtualenv
How to use Matplotlib
How to use iptables
How to use numpy
How to use TokyoTechFes2015
How to use venv
How to use Pyenv
How to use list []
How to use python-kabusapi
How to use OptParse
How to use return
Use Enums with SQLAlchemy
How to use dotenv
How to use pyenv-virtualenv
How to use Go.mod
How to use imutils
How to use import
How to use GCP trace with open Telemetry
How to use tkinter with python in pyenv
Connect to BigQuery with Python
How to use Qt Designer
How to use search sorted
How to use xgboost: Multi-class classification with iris data
python3: How to use bottle (2)
Understand how to use django-filter
How to use the generator
Connect to Wikipedia with Python
Use DATE_FORMAT with SQLAlchemy filter
How to convert a class object to a dictionary with SQLAlchemy