使用 sqlalchemy ，一般通过 Session 对象 ORM 方式操作数据库。如果需要通过 原生 SQL 语句操作数据库，就需要跟 Engine 和 Connect 对象打交道。
Engine 对象包含数据库连接池和数据库方言，通过 create_engine() 函数来创建，engine 对象的 connect() 方法返回 Connection 对象，Connection 对象提供 execute() 方法，允许通过原生 sql 语句来操作数据库。
本篇以 SQLite 数据库为例，简单介绍原生 SQL 操作的一些要点。
-
create_engine() 函数创建 Engine 对象
-
Engine.connect() 方法返回 Connection 对象
-
Connection.execute() 方法执行 SQL 语句
-
Connection.close() 方法释放资源
from sqlalchemy import create_engine
engine = create_engine('sqlite:///testdb.db')
def test_select_statement():
with engine.connect() as conn:
result_proxy = conn.execute("select * from employees")
result = result_proxy.fetchall()
for item in result:
print(item)
SQLAlchemy 支持两种格式的 sql 语句：？和 :number。
def test_parameter_method1():
with engine.connect() as conn:
conn.execute(
"""INSERT INTO employees
(EMP_ID, FIRST_NAME, LAST_NAME, GENDER,
AGE, EMAIL, PHONE_NR,EDUCATION,
MARITAL_STAT, NR_OF_CHILDREN)
VALUES (?,?,?,?,?,?,?,?,?,?);
""",
('9002', 'Stone2', 'Wang', 'M', 20,
'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0)
方法 2 :
def test_parameter_method2():
with engine.connect() as conn:
conn.execute(
"""INSERT INTO employees
(EMP_ID, FIRST_NAME, LAST_NAME, GENDER,
AGE, EMAIL, PHONE_NR, EDUCATION,
MARITAL_STAT, NR_OF_CHILDREN)
VALUES (:1,:2,:3,:4,:5,:6,:7,:8,:9,:10);
""",
('9003', 'Stone3', 'Wang', 'M', 20, 'stone@gmail.com',
'138xxx', 'Bachelor', 'Single', 0)
SQLAlchemy 支持一次插入多行，需要插入的数据放在 list 中：
def test_insert_multiple_rows(self):
with engine.connect() as conn:
values = [
('9004', 'Stone4', 'Wang', 'M', 20, 'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0),
('9005', 'Stone5', 'Wang', 'M', 20, 'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0),
('9006', 'Stone6', 'Wang', 'M', 20, 'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0)
conn.execute(
"""INSERT INTO employees
(EMP_ID, FIRST_NAME, LAST_NAME, GENDER,
AGE, EMAIL, PHONE_NR, EDUCATION,
MARITAL_STAT, NR_OF_CHILDREN)
VALUES (?,?,?,?,?,?,?,?,?,?);
""", values)
由于执行 sql 插入操作自动提交 (commit)，sqlalchemy 提供了 Transactions 来管理 commit 和 rollback，需要提交的时候用 commit() 方法，需要回滚的时候用 rollback() 方法。
def test_txn(self):
conn = engine.connect()
with conn.begin() as txn:
conn.execute(
"""INSERT INTO employees
(EMP_ID, FIRST_NAME, LAST_NAME, GENDER, AGE, EMAIL, PHONE_NR,
EDUCATION, MARITAL_STAT, NR_OF_CHILDREN)
VALUES (:1,:2,:3,:4,:5,:6,:7,:8,:9,:10);
""",
('9007', 'Stone7', 'Wang', 'M', 20, 'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0)
txn.commit()
conn.close()
github - executing raw sql statement
使用 sqlalchemy ，一般通过 Session 对象 ORM 方式操作数据库。如果需要通过 原生 SQL 语句操作数据库，就需要跟 Engine 和 Connect 对象打交道。Engine 对象包含数据库连接池和数据库方言，通过 create_engine() 函数来创建，engine 对象的 connect() 方法返回 Connection 对象，Connection 对象提供 execute() 方法，允许通过原生 sql 语句来操作数据库。本篇以 SQLite 数据库为例，简单介绍原生
选用sqlalchemy+mysqlconnector，连接数据库，创建表，对指定表进行CRUD
from sqlalchemy import exists, Column, Integer, String, ForeignKey, DateTime, Text, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalche
SQLAlchemy很棒，但是可能很难设置。 使用SQLA-Wrapper，您可以像这样快速启动：
from sqla_wrapper import SQLAlchemy
db = SQLAlchemy ( 'sqlite:///:memory:' )
class User ( db . Model ):
__tablename__ "users"
id = db . Column ( db . Integer , primary_key = True )
db . create_all ()
一、问题：有时候 ，我们希望通过原生 sql 来查询数据库，这个时候 应该怎么办呢？
sqlalchemy 已经给我们提供了这样的接口， 通过 text 就可以轻松实现了。
二、代码实现：
1.无参数查询：
# 无参数查询
sql = 'SELECT machine_room_name,type FROM machine_room limit 2'
result = session.execute(text(sql))
print(type(r
cursor = session.execute('insert into users(name) values(:value)', params={"value": 'abc'})
se...
cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'})
session.commit()
print(cursor.lastrowid)
代码如下:select to_char(date_published, ‘yyyymm’) yo from blog group by yo; 转换成sqlalchemy 语句
可以使用 extract() 代码如下:In[3]: year_field = db.func.extract(‘year’, ArticleModel.date_published) In[4]: month_field = db.func.extract(‘month’, ArticleModel.date_published) In[5]: yonth_field = year_field *100+ month
sql to sqlalchemy实例教程-使用MySQL示例雇员数据库
在Python项目中，经常需要操作数据库，而sqlalchemy提供了SQL工具包和对象关系映射（ORM）工具，大大提高了编程开发的效率。为了更好地提升自己的sql以及使用sqlachemy水平，可以使用MySQL自带的示范数据库员工进行练习。
未完待续... ，敬请期待后续的复杂的联合查询语句。
欢迎大家提供需要转变为sqlalchemy语法的sql语句。
如有需求，请加入QQ群：291521082
class demo
class DbBase:
def __init__(self, db_choice: str, db_user: str, db_password: str, host: str, port: int, db_name: str,
pool_size=int(SQLALCHEMY_POOL_SIZE), max_overflow=int(SQLALCHEMY_POOL_MAX_SIZE),
pool_recy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from ...
目录安装导入sqlalchemy执行流程创建引擎执行sql方式一方式二两种方式的区别增删改查获取结果fetchonefetchmanyfetchall
pip3 install -i http://pypi.douban.com/simple pymysql sqlalchemy --trusted-host pypi.douban.com
import sqlalchemy
sql...
假如你的项目有需要将orm对象转为
sql语句去拼接的需求，可以试试下面这种方法
通常要将orm对象转为
sql，可以直接str(对象)，但出来的
sql在条件值的地方可能不是你想要的，比如：
orm对象是：
str()转出来的是：
这肯定是不符合我们的需求的
用下面这种方式可以转出源生
sql，具体原理自己多研究下
from
sqlalchemy.dialects import postgre
sql
# 2.给定ORM Query对象时，为了获得该ClauseElement.compile()方法，需要statement首先访问访问者：
statement = query.statement
# 3.使用literal_binds标志，传递给compile_kwargs来获取参数：
SQLAlchemy有三种方式可以执行原生SQL语句：
1. 使用execute()方法：可以直接在SQLAlchemy中执行任意SQL语句，无需使用ORM模型。例如：
```python
from sqlalchemy import create_engine
engine = create_engine('postgresql://user:password@hostname/database')
with engine.connect() as connection:
result = connection.execute("SELECT * FROM my_table")
2. 使用text()方法：使用text()方法可以将SQL语句转换为SQLAlchemy的文本对象，这个对象可以被用于SQLAlchemy的查询语句中。例如：
```python
from sqlalchemy import text
from myapp import db
result = db.session.query(User).filter(text("name = 'John'")).all()
3. 使用 raw SQL 得到 结果集 ：
```python
from sqlalchemy.orm import sessionmaker
from myapp import engine, User
Session = sessionmaker(bind=engine)
session = Session()
result = session.execute('SELECT * FROM users')
for row in result:
print(row)
