Essential_SQLAlchemy 2 th学習ノートのCoreモジュール

SQL Expression LanguageはオリジナルSQL言語を簡単にパッケージ化した2大モジュールSQLAlchemy Core and ORM:

Core:SQL Expression Languageを実行するインタフェース

を提供する

ORM

インストール:SQLAlchemyおよび関連データベースドライバpip install sqlalchemy pymysql データベースへの接続
データベース接続文字列フォーマット:ここを参照してください

mysql://username:password@hostname/database
postgresql://username:password@hostname/database
sqlite:////absolute/path/to/database
oracle://scott:[email protected]:1521/orcl

例えばSQLiteは次のようになります.

from sqlalchemy import create_engine
engine = create_engine('sqlite:///cookies.db')
engine2 = create_engine('sqlite:///:memory:')
engine3 = create_engine('sqlite:////home/cookiemonster/cookies.db')
engine4 = create_engine('sqlite:///c:\\Users\\cookiemonster\\cookies.db')

：create_engine engine ， ， engine connection MySQL 8 ： pool_recycle=3600

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://cookiemonster:[email protected]/cookies', pool_recycle=3600)

create_Engineの残りのパラメータ:

echo:実行するsql文とそのパラメータをlog印刷するかどうか.デフォルトはFalse

encoding:デフォルトutf-8

isolation_レベル:独立性レベル

pool_recycle:接続回収間隔を指定します.これはMySQL接続の8時間メカニズムにとって特に重要です.デフォルト-1

接続の取得

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://cookiemonster:chocolatechip' \
'@mysql01.monster.internal/cookies', pool_recycle=3600)
connection = engine.connect()

Schema and Types
•Generic•SQL Standard•Vendor specific•User definedの4種類のコレクション
SQLAlchemyは、異なるデータベースと互換性を持つために多くのgeneric typesを定義しています.これらのタイプはsqlalchemy.typesモジュールに定義されており、sqlalchemyから直接インポートすることもできます.タイプ対応表は次のとおりです.
SQLAlchemy
Python
SQL
BigInteger
int
BIGINT
Boolean
bool
BOOLEAN or SMALLINT
Date
datetime.date
DATE (SQLite: STRING)
DateTime
datetime.datetime
DATETIME (SQLite: STRING)
Enum
str
ENUM or VARCHAR
Float
float or Decimal
FLOAT or REAL
Integer
int
INTEGER
Interval
datetime.timedelta
INTERVAL or DATE from epoch
LargeBinary
byte
BLOB or BYTEA
Numeric
decimal.Decimal
NUMERIC or DECIMAL
Unicode
unicode
UNICODE or VARCHAR
Text
str
CLOB or TEXT
Time
datetime.time
DATETIME
これらのタイプがjsonタイプをサポートするデータベースなど、あなたを満たすことができない場合は、sqlalchemy.dialectsモジュールの対応するデータベースのタイプを使用する必要があります.例えばfrom sqlalchemy.dialects.postgresql import JSONMetadata & Table & Column
Metadataは、データベースへの迅速なアクセスを目的としています.多くのTableオブジェクトの集合と見なすことができ,engin,connectionに関する情報もいくつかある.MetaData.tablesでこれらの表オブジェクト辞書にアクセスして、表オブジェクトを定義する前にMetadataをインスタンス化する必要があります.

from sqlalchemy import MetaData
metadata = MetaData()

Tableオブジェクトは、1番目のパラメータが名前、2番目のパラメータがMetadataオブジェクト、その後のパラメータがColumnオブジェクトとして構築する.Columnオブジェクトのパラメータは、名前、タイプ、およびその他です.

from sqlalchemy import Table, Column, Integer, Numeric, String, ForeignKey
cookies = Table('cookies', metadata,
Column('cookie_id', Integer(), primary_key=True),
Column('cookie_name', String(50), index=True),
Column('cookie_recipe_url', String(255)),
Column('cookie_sku', String(55)),
Column('quantity', Integer()),
Column('unit_cost', Numeric(12, 2))
)

from datetime import datetime
from sqlalchemy import DateTime
users = Table('users', metadata,
Column('user_id', Integer(), primary_key=True),
Column('username', String(15), nullable=False, unique=True),
Column('email_address', String(255), nullable=False),
Column('phone', String(20), nullable=False),
Column('password', String(25), nullable=False),
Column('created_on', DateTime(), default=datetime.now),
Column('updated_on', DateTime(), default=datetime.now, onupdate=datetime.now)

注意:ここでdefault、onupdate属性は直接値ではなくcallableオブジェクトです.例えばdatetime.now()は、インスタンスごとにインスタンス化、更新される時間ではなく、永遠にこの値であるからです.比較的有用なのはonupdateであり、更新のたびにメソッドまたは関数が呼び出されます.
キーとコンストレイント(Keys and Constraints)キーとコンストレイントは、上述したようにkwargsによってColumnに定義してもよいし、後でオブジェクトによって追加してもよい.関連クラスは、最も一般的な3つのfrom sqlalchemy import PrimaryKeyConstraint, UniqueConstraint, CheckConstraintなどのベースのsqlalchemyモジュールで定義されます.

PrimaryKeyConstraint('user_id', name='user_pk'),                。
UniqueConstraint('username', name='uix_username')
CheckConstraint('unit_cost >= 0.00', name='unit_cost_positive')

インデックス(Index)

from sqlalchemy import Index
Index('ix_cookies_cookie_name', 'cookie_name')

この定義はTableコンストラクタに配置する必要があります.Index('ix_test', mytable.c.cookie_sku, mytable.c.cookie_name))のように、後に定義することもできます.
関連関係と外部キー制約(Relationships and ForeignKeyConstraints)

from sqlalchemy import ForeignKey
orders = Table('orders', metadata,
Column('order_id', Integer(), primary_key=True),
Column('user_id', ForeignKey('users.user_id')),
Column('shipped', Boolean(), default=False)
)
line_items = Table('line_items', metadata,
Column('line_items_id', Integer(), primary_key=True),
Column('order_id', ForeignKey('orders.order_id')),
Column('cookie_id', ForeignKey('cookies.cookie_id')),
Column('quantity', Integer()),
Column('extended_cost', Numeric(12, 2))
)

注意:ここでForeignKeyは、参照ではなく、データベース内のテーブル名.カラム名に対応する文字列パラメータを使用します.モジュール間の相互依存性を分離しても使用できます:ForeignKeyConstraint(['order_id'], ['orders.order_id'])テーブル・モードの作成または永続化(Persisting the Table)サンプル・コードでは、すべてのTable定義と、追加のモード定義がmetadataオブジェクトに関連付けられていることがわかります.このmetadataオブジェクトを使用してテーブルを作成できます.

metadata.create_all(engine)

注:デフォルトではcreate_all は既存のテーブルを再作成するので、安全に複数回呼び出すことができ、Ablembicなどのデータベース移行ライブラリとの統合も非常に友好的で、追加の手動符号化を必要としません.
このセクションのコードは次のとおりです.

from datetime import datetime
from sqlalchemy import (MetaData, Table, Column, Integer, Numeric, String,
DateTime, ForeignKey, create_engine)
metadata = MetaData()
cookies = Table('cookies', metadata,
Column('cookie_id', Integer(), primary_key=True),
Column('cookie_name', String(50), index=True),
Column('cookie_recipe_url', String(255)),
Column('cookie_sku', String(55)),
Column('quantity', Integer()),
Column('unit_cost', Numeric(12, 2))
)
users = Table('users', metadata,
Column('user_id', Integer(), primary_key=True),
Column('customer_number', Integer(), autoincrement=True),
Column('username', String(15), nullable=False, unique=True),
Column('email_address', String(255), nullable=False),
Column('phone', String(20), nullable=False),
Column('password', String(25), nullable=False),
Column('created_on', DateTime(), default=datetime.now),
Column('updated_on', DateTime(), default=datetime.now, onupdate=datetime.now)
)
orders = Table('orders', metadata,
Column('order_id', Integer(), primary_key=True),
Column('user_id', ForeignKey('users.user_id'))
)
line_items = Table('line_items', metadata,
Column('line_items_id', Integer(), primary_key=True),
Column('order_id', ForeignKey('orders.order_id')),
Column('cookie_id', ForeignKey('cookies.cookie_id')),
Column('quantity', Integer()),
Column('extended_cost', Numeric(12, 2))
)
engine = create_engine('sqlite:///:memory:')
metadata.create_all(engine)

SQLAlchemy-Coreモジュール
データの挿入:

ins = cookies.insert().values(
cookie_name="chocolate chip",
cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
cookie_sku="CC01",
quantity="12",
unit_cost="0.50"
)
print(str(ins))

もちろんあなたもそうすることができます.

from sqlalchemy import insert
ins = insert(cookies).values(
cookie_name="chocolate chip",
cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
cookie_sku="CC01",
quantity="12",
unit_cost="0.50"
)

上記のプリコンパイル文は次のとおりです.

INSERT INTO cookies
(cookie_name, cookie_recipe_url, cookie_sku, quantity, unit_cost)
VALUES
(:cookie_name, :cookie_recipe_url, :cookie_sku, :quantity, :unit_cost)

実際のプロシージャは、insオブジェクトの内部でcompile()メソッドを呼び出して上記の文にコンパイルし、パラメータをinsに格納する.compile().params辞書にあります.次に、前に取得したconnectionオブジェクトからstatementを実行します.

result = connection.execute(ins)

もちろん、このように調べることもできます.

ins = cookies.insert()
result = connection.execute(
ins,
cookie_name='dark chocolate chip',
cookie_recipe_url='http://some.aweso.me/cookie/recipe_dark.html',
cookie_sku='CC02',
quantity='1',
unit_cost='0.75'
)
result.inserted_primary_key

一括挿入:

inventory_list = [
{
'cookie_name': 'peanut butter',
'cookie_recipe_url': 'http://some.aweso.me/cookie/peanut.html',
'cookie_sku': 'PB01',
'quantity': '24',
'unit_cost': '0.25'
},
{
'cookie_name': 'oatmeal raisin',
'cookie_recipe_url': 'http://some.okay.me/cookie/raisin.html',
'cookie_sku': 'EWW01',
'quantity': '100',
'unit_cost': '1.00'
}
]
result = connection.execute(ins, inventory_list)

注意:すべての辞書パラメータが同じkeysを持っていることを確認してください.
検索

from sqlalchemy.sql import select
s = select([cookies])
rp = connection.execute(s)
results = rp.fetchall()

もちろん、文字列の代わりに文字列を使用することもできます.

s = select("""SELECT cookies.cookie_id, cookies.cookie_name,
cookies.cookie_recipe_url, cookies.cookie_sku, cookies.quantity,
cookies.unit_cost FROM cookies""")

connection.executeが返すrp変数はResultProxyオブジェクト(DBAPIのcursorオブジェクトのカプセル化)です.
私たちもこのように書くことができます.

from sqlalchemy.sql import select
s = cookies.select()
rp = connection.execute(s)
results = rp.fetchall()

ResultProxyにより、クエリ結果はindex,name,or Column objectを介してカラムデータにアクセスできます.例:

first_row = results[0]
first_row[1] #      1  ,by index
first_row.cookie_name # by name
first_row[cookies.c.cookie_name] #by Column object.

ResultProxyを反復することもできます.次のようにします.

rp = connection.execute(s)
for record in rp:
print(record.cookie_name)

ResultProxyの残りの結果セットを取得する方法

first()

fetchone()

fetchall()

scalar():Returns a single value if a query results in a single record with one column.

keys()カラム名

を取得
ResultProxyの上記の方法の選択についてのアドバイス:1、fetchone()ではなくfirst()を使用して単一のレコードを取得する.fetchone()呼び出し後も開いているconnectionsが続いているため、うっかりすると問題を起こしやすい.2、fetchall()ではなく反復方式ですべての結果を取得し、メモリをより節約します.3、scalar()を使用して単行単列の結果を取得する場合は、複数の行を返すと例外が放出されることに注意してください.
戻りカラム数の制御

s = select([cookies.c.cookie_name, cookies.c.quantity])
rp = connection.execute(s)
print(rp.keys())
result = rp.first()

ツールバーの

s = select([cookies.c.cookie_name, cookies.c.quantity])
s = s.order_by(cookies.c.quantity)
rp = connection.execute(s)
for cookie in rp:
print('{} - {}'.format(cookie.quantity, cookie.cookie_name))

#  desc
from sqlalchemy import desc
s = select([cookies.c.cookie_name, cookies.c.quantity])
s = s.order_by(desc(cookies.c.quantity))

結果セットを返すバーの数を制限

s = select([cookies.c.cookie_name, cookies.c.quantity])
s = s.order_by(cookies.c.quantity)
s = s.limit(2)
rp = connection.execute(s)
print([result.cookie_name for result in rp])

内蔵SQL関数
ここにいるよsql.funcモジュール内

#sum
from sqlalchemy.sql import func
s = select([func.sum(cookies.c.quantity)])
rp = connection.execute(s)
print(rp.scalar())

#count
s = select([func.count(cookies.c.cookie_name)])
rp = connection.execute(s)
record = rp.first()
print(record.keys())
print(record.count_1) #         ，_，       ，   

#    
s = select([func.count(cookies.c.cookie_name).label('inventory_count')])
rp = connection.execute(s)
record = rp.first()
print(record.keys())
print(record.inventory_count)

フィルタリング

#where
s = select([cookies]).where(cookies.c.cookie_name == 'chocolate chip')
rp = connection.execute(s)
record = rp.first()
print(record.items()) #  row   items()  。

#like
s = select([cookies]).where(cookies.c.cookie_name.like('%chocolate%'))
rp = connection.execute(s)
for record in rp.fetchall():
    print(record.cookie_name)

whereで使用できる句要素

between(cleft, cright)

concat(column_two) Concatenate column with column_two

distinct()

in_([list])

is_(None) Find where the column is None (commonly used for Null checks with None)

contains(string) Find where the column has string in it (case-sensitive)

endswith(string) Find where the column ends with string (case-sensitive)

like(string) Find where the column is like string (case-sensitive)

startswith(string) Find where the column begins with string (case-sensitive)

ilike(string) Find where the column is like string (this is not case-sensitive)

もちろん、notin_のような一連のnotxxxメソッドも含まれています.()、唯一の例外はisnot()
オペレータ

+,-,*,/,%

==,!=,,<=,>=

AND,OR,NOT,pythonキーワードのせいでand_(),or_(),not_()

の代わりに
+記号は、文字列の結合にも使用できます.

s = select([cookies.c.cookie_name, 'SKU-' + cookies.c.cookie_sku])
for row in connection.execute(s):
print(row)

from sqlalchemy import cast
s = select([cookies.c.cookie_name,
    cast((cookies.c.quantity * cookies.c.unit_cost),
        Numeric(12,2)).label('inv_cost')])
for row in connection.execute(s):
    print('{} - {}'.format(row.cookie_name, row.inv_cost))

注意:castは別の関数で、数値を通貨形式に変換するタイプ変換とprint('{}-{:.2f}'.format(row.cokie_name,row.inv_cost))を行うことができる.この行為は一致している.

from sqlalchemy import and_, or_, not_
s = select([cookies]).where(
    and_(
        cookies.c.quantity > 23,
        cookies.c.unit_cost < 0.40
    )
)
for row in connection.execute(s):
    print(row.cookie_name)


from sqlalchemy import and_, or_, not_
s = select([cookies]).where(
    or_(
        cookies.c.quantity.between(10, 50),
        cookies.c.cookie_name.contains('chip')
    )
)
for row in connection.execute(s):
    print(row.cookie_name)

update

from sqlalchemy import update
u = update(cookies).where(cookies.c.cookie_name == "chocolate chip")
u = u.values(quantity=(cookies.c.quantity + 120))
result = connection.execute(u)
print(result.rowcount)
s = select([cookies]).where(cookies.c.cookie_name == "chocolate chip")
result = connection.execute(s).first()
for key in result.keys():
    print('{:>20}: {}'.format(key, result[key]))

delete

from sqlalchemy import delete
u = delete(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
result = connection.execute(u)
print(result.rowcount)

s = select([cookies]).where(cookies.c.cookie_name == "dark chocolate chip")
result = connection.execute(s).fetchall()
print(len(result))

joins
join()、outerjoin()関数、select_from()関数

columns = [orders.c.order_id, users.c.username, users.c.phone,
           cookies.c.cookie_name, line_items.c.quantity,
           line_items.c.extended_cost]
cookiemon_orders = select(columns)
cookiemon_orders = cookiemon_orders.select_from(orders.join(users).join(
    line_items).join(cookies)).where(users.c.username ==
                                     'cookiemon')
result = connection.execute(cookiemon_orders).fetchall()
for row in result:
    print(row)

最終的に生成されたSQL文は次のとおりです.

SELECT orders.order_id, users.username, users.phone, cookies.cookie_name,
line_items.quantity, line_items.extended_cost FROM users JOIN orders ON
users.user_id = orders.user_id JOIN line_items ON orders.order_id =
line_items.order_id JOIN cookies ON cookies.cookie_id = line_items.cookie_id
WHERE users.username = :username_1

outerjoin

columns = [users.c.username, func.count(orders.c.order_id)]
all_orders = select(columns)
all_orders = all_orders.select_from(users.outerjoin(orders))
all_orders = all_orders.group_by(users.c.username)
result = connection.execute(all_orders).fetchall()
for row in result:
    print(row)

テーブル別名関数alias()

>>> manager = employee_table.alias('mgr')
>>> stmt = select([employee_table.c.name],
            ... and_(employee_table.c.manager_id==manager.c.id,
            ... manager.c.name=='Fred'))
>>> print(stmt)
SELECT employee.name
FROM employee, employee AS mgr
WHERE employee.manager_id = mgr.id AND mgr.name = ?

グループ化

columns = [users.c.username, func.count(orders.c.order_id)]
all_orders = select(columns)
all_orders = all_orders.select_from(users.outerjoin(orders))
all_orders = all_orders.group_by(users.c.username)
result = connection.execute(all_orders).fetchall()
for row in result:
    print(row)

chaining

def get_orders_by_customer(cust_name, shipped=None, details=False):
    columns = [orders.c.order_id, users.c.username, users.c.phone]
    joins = users.join(orders)
    if details:
        columns.extend([cookies.c.cookie_name, line_items.c.quantity,
            line_items.c.extended_cost])
        joins = joins.join(line_items).join(cookies)
    cust_orders = select(columns)
    cust_orders = cust_orders.select_from(joins)

    cust_orders = cust_orders.where(users.c.username == cust_name)
    if shipped is not None:
        cust_orders = cust_orders.where(orders.c.shipped == shipped)
    result = connection.execute(cust_orders).fetchall()
    return result

オリジナルSQLの実行
返されるのはResultProxyオブジェクト1で、元のSQLを完全に使用します.

result = connection.execute("select * from orders").fetchall()
print(result)

2、部分的に原始SQL、text()関数を採用する

from sqlalchemy import text
stmt = select([users]).where(text("username='cookiemon'"))
print(connection.execute(stmt).fetchall())

異常
SQLALchemyは多くの異常を定義している.私たちは関心を持っています:AttributeErrors,IntegratyErrors.などの試験と説明を行うために、まず次の文を実行してください.

from datetime import datetime
from sqlalchemy import (MetaData, Table, Column, Integer, Numeric, String,
                        DateTime, ForeignKey, Boolean, create_engine,
                        CheckConstraint)
metadata = MetaData()
cookies = Table('cookies', metadata,
                Column('cookie_id', Integer(), primary_key=True),
                37
                Column('cookie_name', String(50), index=True),
                Column('cookie_recipe_url', String(255)),
                Column('cookie_sku', String(55)),
                Column('quantity', Integer()),
                Column('unit_cost', Numeric(12, 2)),
                CheckConstraint('quantity > 0', name='quantity_positive')
                )
users = Table('users', metadata,
              Column('user_id', Integer(), primary_key=True),
              Column('username', String(15), nullable=False, unique=True),
              Column('email_address', String(255), nullable=False),
              Column('phone', String(20), nullable=False),
              Column('password', String(25), nullable=False),
              Column('created_on', DateTime(), default=datetime.now),
              Column('updated_on', DateTime(),
                     default=datetime.now, onupdate=datetime.now)
              )
orders = Table('orders', metadata,
               Column('order_id', Integer()),
               Column('user_id', ForeignKey('users.user_id')),
               Column('shipped', Boolean(), default=False)
               )
line_items = Table('line_items', metadata,
                   Column('line_items_id', Integer(), primary_key=True),
                   Column('order_id', ForeignKey('orders.order_id')),
                   Column('cookie_id', ForeignKey('cookies.cookie_id')),
                   Column('quantity', Integer()),
                   Column('extended_cost', Numeric(12, 2))
                   )
engine = create_engine('sqlite:///:memory:')
metadata.create_all(engine)
connection = engine.connect()

from sqlalchemy import select, insert
ins = insert(users).values(
username="cookiemon",
email_address="[email protected]",
phone="111-111-1111",
password="password"
)
result = connection.execute(ins)
s = select([users.c.username])
results = connection.execute(s)
for result in results:
print(result.username)
print(result.password) #   AttributeError  

IntegratyError異常は、制約に違反した場合に発生します.例えば、一意性の制約に違反するなどです.

s = select([users.c.username])
connection.execute(s).fetchall()
[(u'cookiemon',)]
ins = insert(users).values(
    username="cookiemon",
    email_address="[email protected]",
    phone="111-111-1111",
    password="password"
)
result = connection.execute(ins) #   IntegrityError, UNIQUE constraint failed: users.username
#    
try:
    result = connection.execute(ins)
except IntegrityError as error:
    print(error.orig.message, error.params)

すべてのSQLAlchemy異常処理方式は上記の考え方であり,[SQLAlchemyError](http://docs.sqlalchemy.org/en/latest/core/exceptions.html)で取得できる情報は次のとおりです.

orig :The DBAPI exception object.

params:The parameter list being used when this exception occurred.

statement :The string SQL statement being invoked when this exception occurred.

トランザクションTransactions

from sqlalchemy.exc import IntegrityError


def ship_it(order_id):
    s = select([line_items.c.cookie_id, line_items.c.quantity])
    s = s.where(line_items.c.order_id == order_id)
    transaction = connection.begin() #    
    cookies_to_ship = connection.execute(s).fetchall()
    try:
        for cookie in cookies_to_ship:
            u = update(cookies).where(cookies.c.cookie_id == cookie.cookie_id)
            u = u.values(quantity=cookies.c.quantity - cookie.quantity)
            result = connection.execute(u)
        u = update(orders).where(orders.c.order_id == order_id)
        u = u.values(shipped=True)
        result = connection.execute(u)
        print("Shipped order ID: {}".format(order_id))
        transaction.commit() #    
    except IntegrityError as error:
        transaction.rollback()   #    
        print(error)