SqlAlchemy使用教程(四) MetaData 与 SQL Express Language 的使用

本文介绍: MetaData 相当于python层的db schema，即数据库结构定义, 用meta.Table对象来表示table 定义，Column对象来表示 column的定义，通常1个模块只包含1个metaData对象，可以包含多个table定义。Index。

在这里插入图片描述

MetaData对象用于描述表结构，SQL Express Language是DBAPI SQL的统一封装器。MetaData 与SQL Express 语句可以在Core层使用，ORM层基于MetaData, SQL Express基础上做了进一步抽象。本章将介绍在Core层如何使用MetaData与SQL Express Language语句。

from sqlalchemy import MetaData
metadata_obj = MetaData()

创建了MetaData对象后，就可以用它来声明Table对象，每个字段用Column对象来表示

user_table = Table(
    'user_account',
    metadata_obj,
    Column('id',Integer,primary_key=True),
    Column('name',String(30)),
    Column('speciality',String(30)),
)

说明：

MetaData对象column对象来表示数据库字段，其主要属性

table对象的 c属性 , 即 table.c
所有列名被放进 table.c数组中，引用列名方式：user_table.c.name

Column("id", Integer, primary_key=True),

Column(‘Addres’, String, index=True)

Column("user_id", ForeignKey("user_account.id"), nullable=False),

metadata_obj.create_all()

print(meta_obj.tables) 
user_account_tbl = meta_obj.tables['user_account']

meta_obj.drop_all()

此功能通过 metadata类的reflect() 方法实现。

engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test_db?charset=utf8")
meta_obj = MetaData() 
meta_obj.reflect(bind=engine) 
school_table = meta_obj.tables[‘school’]

>>> messages = Table("messages", metadata_obj, autoload_with=engine)
>>> [c.name for c in messages.columns]
['message_id', 'message_name', 'date']

>>> shopping_cart_items = Table("shopping_cart_items", metadata_obj, autoload_with=engine)
>>> "shopping_carts" in metadata_obj.tables
True

with engine.connect() as conn: 
    stmt = insert(asset_table).values(name="打印机",tag="A0001",value=3000,user_id=1)
    result = conn.execute(stmt)
    conn.commit()
    print(result.inserted_primary_key)

with engine.connect() as conn:     
    # send many statements 
    rows = [
        {'name':'复印机','tag': 'A0002', 'value': 29000, 'user_id': 4 },
        {'name':'20吨吊车','tag': 'D0001', 'value': 240000, 'user_id': 1 },
    ]
    conn.execute( asset_table.insert(), rows )
    conn.commit()

from sqlalchemy import select
stmt = select(user_table).where(user_table.c.name == "spongebob")
print(stmt)
with Session(engine) as session:
     for row in session.execute(stmt):
         print(row)

select(user_table.c.name, user_table.c.fullname))

from sqlalchemy import func, cast
stmt = select(("Username: " + user_table.c.name).label("username"),).order_by(user_table.c.name)
with engine.connect() as conn:
    for row in conn.execute(stmt):
         print(f"{row.username}")

Username: 张锋
Username: 海绵宝宝
Username: 王小乙

select(user_table).where(user_table.c.name == "squidward"))

>>> print(
...     select(address_table.c.email_address)
...     .where(user_table.c.name == "squidward")
...     .where(address_table.c.user_id == user_table.c.id)
... )

SELECT address.email_address
FROM address, user_account
WHERE user_account.name = :name_1 AND address.user_id = user_account.id

>>> print(
...     select(address_table.c.email_address)
...     .select_from(user_table)
...     .join(address_table, user_table.c.id == address_table.c.user_id)
... )

SELECT address.email_address
FROM user_account JOIN address ON user_account.id = address.user_id

stmt = update(user_table).values(fullname="Username: " + user_table.c.name)

>>> from sqlalchemy import bindparam
>>> stmt = (
...     update(user_table)
...     .where(user_table.c.name == bindparam("oldname"))
...     .values(name=bindparam("newname"))
... )
>>> with engine.begin() as conn:
...     conn.execute(
...         stmt,
...         [
...             {"oldname": "jack", "newname": "ed"},
...             {"oldname": "wendy", "newname": "mary"},
...             {"oldname": "jim", "newname": "jake"},
...         ],
...     )

>>> scalar_subq = (
...     select(address_table.c.email_address)
...     .where(address_table.c.user_id == user_table.c.id)
...     .order_by(address_table.c.id)
...     .limit(1)
...     .scalar_subquery()
... )
>>> update_stmt = update(user_table).values(fullname=scalar_subq)
>>> print(update_stmt)

from sqlalchemy import delete
stmt = (
    delete(user_table).
    where(user_table.c.id == 5)
)
result = conn.execute(stmt)

delete_stmt = (
     delete(user_table)
     .where(user_table.c.id == address_table.c.user_id)
     .where(address_table.c.email_address == "patrick@aol.com")
)
from sqlalchemy.dialects import mysql
print(delete_stmt.compile(dialect=mysql.dialect()))

DELETE FROM user_account USING user_account, address
WHERE user_account.id = address.user_id AND address.email_address = %s

显示所有内容

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

四、Database MetaData 与 SQL Express Language 的使用

1、使用MetaData定义表结构

MetaData的含义

Step-1, 创建1个MetaData对象

Step-2 申明 Table对象

Step-3 Columns 对象

Step-4: 定义Primary key、index, foreign key

Step-5: 发送 DDL 指定到数据库创建表

根据已存在数据库反向生成 metadata Table对象

2、SQL Express Language 使用

2.1 NSERT() 方法

2.2 Select()方法

2.3 数据更新update()与删除 Delete()

1) 更新数据

2) 删除数据

发表回复取消回复