Ceci est une note SQLAlchemy qui vous permet d'écrire du code de type python autour de sql. En introduisant une telle bibliothèque, il est possible d'éliminer les erreurs de syntaxe SQL et de créer un service avec peu de bogues.
L'explication ci-dessous est basée sur la version 1.1.
SQLAlchemy permet de gérer les informations de colonne par programmation en définissant une table dans le code. Vous pouvez également transférer la responsabilité de créer la table du manuel au 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)
)
Comme indiqué dans AUTO_INCREMENT Behavior, l'auto-incrémentation est automatiquement ajoutée à la première colonne qui remplit les conditions suivantes. Sera fait.
--clé primaire --Entier --Pas de clé étrangère
Utilisez sqlalchemy.dialects.mysql.INTEGER. dialecte: C'est un dialecte. dialecte mysql. Si vous souhaitez utiliser différents mysql, il peut être préférable d'utiliser le type sous dialects.mysql.
import sqlalchemy as sa
from sqlalchemy.dialects.mysql import INTEGER
...
sa.Column('scenario_id', INTEGER(unsigned=True), nullable=False)
D'autres sont comme dans 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
Etc.
Il existe deux façons de spécifier «default» et «server_default».
--default: donne la valeur par défaut si aucune valeur n'est spécifiée dans la couche python
Ici, nous allons expliquer 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()
La valeur donnée comme dans l'exemple ci-dessus est entre guillemets. Si vous ne voulez pas le citer, utilisez du texte.
Dans le cas habituel de create_datetime, la valeur par défaut est la suivante.
import sqlalchemy as sa
from sqlalchemy.dialects.mysql import DATETIME
sa.Column('create_datetime', DATETIME(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
Comme par défaut, il existe onupdate et server_onupdate. La différence est la même que celle par défaut, utilisez server_onupdate pour modifier l'instruction de création de table.
Un horodatage commun (heure de mise à jour) peut être écrit comme suit.
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'))
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)
)
Supposons qu'une variable appelée «table» soit définie comme ci-dessus.
Nous faisons juste une requête ici, donc cela ne fonctionnera que si nous la mettons réellement dans le moteur de base de données. Veuillez attraper d'autres documents pour savoir comment les insérer. (Je ne l'écrirai pas ici car il semble que ce soit différent de l'usage général car j'utilise 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)
Vous pouvez spécifier la condition avec la liste des colonnes que vous souhaitez insérer dans select (), suivie de where. table.c. représente une colonne. Cela signifie que la valeur de la colonne user_id est 1234.
Il semble que le nom de la colonne ne puisse pas être spécifié dans table.select ().
Lisez ↓ pour plus de détails. http://docs.sqlalchemy.org/en/rel_1_1/core/selectable.html
En passant, si vous souhaitez utiliser les paramètres assignés dans le test, vous pouvez procéder comme suit.
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 est une incrémentation automatique
user_name='hal',
#Puisque l'email n'est pas spécifié, il sera nul
password='greatpassword'
)
Vous pouvez créer une requête avec des valeurs spécifiées par insert (). Values (). c'est facile.
Si vous souhaitez utiliser les paramètres assignés, etc. dans le test, vous pouvez procéder comme suit.
print(str(q))
# INSERT INTO user (user_name, password) VALUES (:user_name, :password)
print(q.compile().params)
# {'user_name': 'hal', 'password': 'greatpassword'}
Voici comment utiliser aiomysql pour exécuter la requête créée ci-dessus. ** Il est courant de créer un moteur parmi les fonctions de sqlalchemy **, donc si vous voulez l'utiliser normalement, lisez ici. Je pense que c'est bon.
Nous utilisons aiomysql avec asyncio, qui est une nouvelle spécification qui fait bien le traitement asynchrone, donc je vais brièvement aborder comment l'utiliser dans ce cas. La documentation officielle est http://aiomysql.readthedocs.io/en/latest/sa.html.
Il est pratique de créer quelque chose avec le gestionnaire de contexte comme indiqué ci-dessous.
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( #Créer un moteur avec aiomysql
host=config['host'], #L'argument peut être celui de aiomysql connect
port=config['port'],
user=config['user'],
password=config['password'],
db=config['database'],
charset='utf8',
autocommit=True, #Si cette valeur est définie sur True, elle sera immédiatement reflétée lorsque la commande d'insertion est exécutée.
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)
Côté utilisateur
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