设计建筑,seo学院培训班,补习班,专业苏州网站建设公司排名SqlAlchemy使用教程(一) 原理与环境搭建SqlAlchemy使用教程(二) 入门示例及编程步骤 三、使用Core API访问与操作数据库
Sqlalchemy 的Core部分集成了DB API, 事务管理#xff0c;schema描述等功能#xff0c;ORM构筑于其上。本章介绍创建 Engine对象#xff0c;使用基本的… SqlAlchemy使用教程(一) 原理与环境搭建SqlAlchemy使用教程(二) 入门示例及编程步骤 三、使用Core API访问与操作数据库
Sqlalchemy 的Core部分集成了DB API, 事务管理schema描述等功能ORM构筑于其上。本章介绍创建 Engine对象使用基本的 Sql Express Language 方法以及如何实现对数据库的CRUD操作等内容。
1、创建DB engine 对象
1.1创建database engine 对象
Engine 是db连接管理类 语法:
from sqlalchemy import create_engine
创建引擎对象
engine create_engine(sqlite:///:memory:, echoTrue)
连接数据库
conn engine.connect()Sqlalchemy.create_engine( ) 方法第1个参数是db连接表达式格式为
dialect[driver]://user:passwordhost/dbnamedialect 通常为数据库类型如sqlite, mysql, mongodb etc.driver 是python 访问数据库的包。 如 sqlitesqlite3, mysqlmysqlconnector
1.2 连接至各类数据库的配置
1.2.1 sqlite 连接
上面示例是sqlite的连接表达式。 Driver是python访问数据库的DBAPI库。
e create_engine(sqlite:///path/to/database.db)如果是绝对地址 sqlite:usr/local/myproject/database.db
:memory 表示使用内存数据库不保存在硬盘。 对于windows 系统
e create_engine(sqlite:///C:\\myapp\\db\\main.db)1.2.2 连接mysql
Mysql 的DBAPI常用的有PyMysql 与 mysql-connector其连接表达式分别为
mysqlpymysql://root:123456192.168.99.240:3306/testdb
mysqlmysqlconnector://roprot:123456192.168.99.240:3306/testdb1.2.3 连接PostgreSQL
通常使用的接口库为 psycopg2
postgresqlpsycopg2://user:passwordhost:port/dbname[?keyvaluekeyvalue...]engine create_engine(postgresqlpsycopg2://scott:tigerlocalhost/test,isolation_levelSERIALIZABLE,
)Ssl连接
engine sa.create_engine(postgresqlpsycopg2://scott:tiger192.168.0.199:5432/test?sslmoderequire
)1.2.4 连接MongoDB
engine create_engine(mongodb:///?ServerMyServerPort27017DatabasetestUsertestPasswordPassword)定义1个mapping类
base declarative_base()
class restaurants(base):
__tablename__ restaurants
borough Column(String,primary_keyTrue)
cuisine Column(String)查询
enginecreate_engine(mongodb:///?ServerMyServerPort27017DatabasetestUsertestPasswordPassword)
factory sessionmaker(bindengine)
session factory()
for instance in session.query(restaurants).filter_by(NameMorris Park Bake Shop):
print(borough: , instance.borough)
print(cuisine: , instance.cuisine)
print(---------)1.3创建connect 对象
语法
conn engine.connect() 如
e create_engine(sqlite:///C:\\myapp\\db\\main.db)
conn e.connect()推荐使用context with 语法使用connect对象
from sqlalchemy import create_engine, text
engine create_engine(sqlite:///C:\\myapp\\db\\main.db)
with engine.connect() as connection:result connection.execute(text(select username from users))for row in result:print(username:, row[username])如果修改了数据应调用 conn.commit() 提交transaction
2. SQL Express Language 常用方法
Sqlalchemy 对sql进行了封装其SQL Express语法比直接使用sql 语句更方便优势是传参与获取返回值更省事。
2.1 使用 text() 生成SQL Express语句
text()方法是CoreAPI中最基础的方法之一主要作用用于封装 sql 语句
from sqlalchemy import textt_sql text(SELECT * FROM users)
result connection.execute(t_sql)传参
t_sql text(SELECT * FROM users WHERE id:user_id)
result connection.execute(t_sql, { ‘user_id’: 12 } )如果使用r” “ 则用 : 来表示:
2.2 bindparams() 方法传参
也可以通过 text(sql_statement).bindparams() 直接构建完整的SQL语句
from sqlalchemy import text, bindparams
stmt text(SELECT id, name FROM user WHERE name:name AND timestamp:timestamp)
stmt stmt.bindparams(namejack,timestampdatetime.datetime(2012, 10, 8, 15, 12, 5)
)
result conn.execute(stmt)
print(result.all())bindparams()中可添加参数Type检查
from sqlalchemy import text
stmt text(SELECT id, name FROM user WHERE name:name AND timestamp:timestamp)
stmt stmt.bindparams(bindparam(name, type_String),bindparam(timestamp, type_DateTime)
)
stmt stmt.bindparams(namejack,timestampdatetime.datetime(2012, 10, 8, 15, 12, 5))
result conn.execute(stmt)
print(result.all())3, 解析查询结果
查询结果类型为 sqlalchemy.engine.Result 类是1个由 object 组成的列表。可以用多种方法访问
all() return all rows in a listcolumns(‘col_1’, ‘col_2’) 指定返回每row 的字段 iterablefetchall(), fetchone(), fetchmany()first() 返回第1行。keys() 返回row的字段名 是iterable 类型mappings(), 列表元素为dict类型result.close() 关闭result对象
说明
遍历查询结果 all()- , fetchall(), fetchmany(), columns 结果为 list[tuple,…], 或iterable,对row 字段 可以用key, index row[0], row[‘id’], row[‘name’], 也可以用row.name 如
result conn.execute(text(select x, y from some_table))
for row in result:print(fRow: {row.x} {row.y})result.mapping() 返回结果的row 类型为dict,
result conn.execute(text(select x, y from some_table))
for dict_row in result.mappings():x dict_row[x]y dict_row[y]4. 使用connect 对象执行CRUD操作
SqlAlchemy可以用connect对象与 session 对象来执行SQL express connect对象是直接调用DBAPI执行SQL语句这是使用SqlAlchemy 最简单的方式,同时支持部分Sqlalchemy 的SQL Express 封装语法但执行的SQL语句依然还要符合各数据库的接口库要求。 Session对象则实现了同1套接口适用于所有数据库。但主要用于ORM API方式。
connect对象操作数据库的好处可使用text()方法生成SQL语句利用bindparams() 传值以及做类型检查。同时支持多线程访问数据库。
创建表的方法前面已讲过。 下面示例为 insert, update, delete 操作
# insert row
print(-*50Insert operation)
stmt text(INSERT INTO some_table VALUES(:x, :y)).bindparams(x6,y19)
with engine.connect() as conn:conn.execute(stmt)conn.commit()result conn.execute( text(select * from some_table) )print(result.all())# update row
print(-*50update operation)
stmt text(UPDATE some_table SET y:y WHERE x:x).bindparams(y99,x5)
with engine.connect() as conn:conn.execute(stmt)conn.commit()result conn.execute( text(select * from some_table) )print(result.all())# delete row
print(-*50delete operation)
stmt text(DELETE FROM some_table WHERE x:x).bindparams(x4)
with engine.connect() as conn:conn.execute(stmt)conn.commit()result conn.execute( text(select * from some_table) )print(result.rowcount)print(result.all())output:
--------------------------------------------------Insert operation
2023-12-03 15:50:36,978 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-12-03 15:50:36,978 INFO sqlalchemy.engine.Engine INSERT INTO some_table VALUES(?, ?)
2023-12-03 15:50:36,978 INFO sqlalchemy.engine.Engine [generated in 0.00085s] (6, 19)
2023-12-03 15:50:36,979 INFO sqlalchemy.engine.Engine COMMIT
2023-12-03 15:50:36,980 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-12-03 15:50:36,980 INFO sqlalchemy.engine.Engine select * from some_table
2023-12-03 15:50:36,981 INFO sqlalchemy.engine.Engine [generated in 0.00132s] ()
[(1, 1), (2, 4), (3, 10), (4, 11), (5, 25), (6, 19)]
2023-12-03 15:50:36,982 INFO sqlalchemy.engine.Engine ROLLBACK
--------------------------------------------------update operation[(1, 1), (2, 4), (3, 10), (4, 11), (5, 99), (6, 19)]
2023-12-03 15:50:36,985 INFO sqlalchemy.engine.Engine ROLLBACK
--------------------------------------------------delete operation
[(1, 1), (2, 4), (3, 10), (5, 99), (6, 19)]
2023-12-03 15:50:36,989 INFO sqlalchemy.engine.Engine ROLLBACK5. 表间关系处理
Sqlalchemy 使用DBAPI处理表间关系语法是依据数据库规定, 但基本均支持标准SQL语法
5.1 创建外键字段的语法 CREATE TABLE tracks(……trackartist INTEGER, -- 外键字段
FOREIGN KEY(trackartist) REFERENCES artist(artistid)
)辅表artist.id字段须为主键或unique index。
5.2 各种表间关系的实现方式
One to one: 还是用 foreign key来实现。One to many: 就是外键Many to many 需要中间表 用2个foreign key 与两张表分别建立 one to many 关系。
示例
import sqlalchemyfrom sqlalchemy import create_engine
from sqlalchemy import text
from sqlalchemy.orm import sessionmaker engine create_engine(sqlite:///order.db)# create table people
with engine.connect() as conn:conn.execute(text(drop table if exists people;))stmt text(CREATE TABLE people(id integer PRIMARY KEY,name TEXT, age INTEGER) )conn.execute(stmt)conn.execute(text(INSERT INTO people (id,name, age) VALUES (:id,:name, :age)),[ {id: 1, name: Jack,age:30 }, {id: 2, name: Smith,age:28 }, {id: 3, name: Wang,age:35 }, ])conn.commit()result conn.execute( text(select * from people) )print(result.rowcount)print(result.all())# create table order
# 创建会话Session
with engine.connect() as conn: conn.execute(text(drop table if exists teams))stmt_1 text(create table teams(id integer PRIMARY KEY,team_name TEXT, pid integer,foreign key (pid) REFERENCES people(id)))conn.execute(stmt_1)conn.commit()conn.execute(text(INSERT INTO teams (id, team_name, pid) VALUES (:id, :team_name, :pid)),[ {id: 101, team_name: TV product,pid:1 }, {id: 102, team_name: Software development,pid:2 }, {id: 103, team_name: Electric development,pid:2 }, ])
conn.commit()# 跨表查询result conn.execute( text(select a.id, a.team_name, b.name from teams as a left join people as b on a.pidb.id) )print(result.rowcount)for row in result.mappings():print(row[id], row[team_name], row[name])6. 通过多线程访问Database
sqlalchemy的engine可做为全局变量 将connect对象,或 session对象传入线程实现多线程访问
示例
def thread_db(conn,name):try: result conn.execute( text(select * from people) )print(result.rowcount)print(fthread {{ name }} result: )print(result.all())except Exception as e:print(cant open connection object)finally: conn.close()from threading import Threadt1 Thread(targetthread_db, args(engine.connect(),thread_a))
t2 Thread(targetthread_db, args(engine.connect(),thread_b))
t1.start()
t2.start()
t1.join()
t2.join()
print(main thread is ended)
output:
thread { name } result:
thread { name } result:
[(1, Jack, 30), (2, Smith, 28), (3, Wang, 35)]
[(1, Jack, 30), (2, Smith, 28), (3, Wang, 35)]
main thread is ended
