Sqlalchemy 快速生成 sql 语句

In [104]:
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import config

SQLALCHEMY_DATABASE_URL = config.DATABASE_URL

engine = create_engine(SQLALCHEMY_DATABASE_URL)
Session = sessionmaker(bind=engine)
session = Session()

Base = declarative_base()

class User(Base):
    __tablename__ = 'test_user'
    id = Column(Integer, primary_key = True)
    name = Column(String, default="")
    score = Column(Integer, default=0)

Table 从句用法

In [99]:
query = User.__table__.select()
print(query)

SELECT test_user.id, test_user.name, test_user.score 
FROM test_user
In [101]:
query = User.__table__.select(whereclause=User.id == 1)
print(query)

SELECT test_user.id, test_user.name, test_user.score 
FROM test_user 
WHERE test_user.id = :id_1
In [102]:
query = User.__table__.select().where(User.id == 1)
print(query)

SELECT test_user.id, test_user.name, test_user.score 
FROM test_user 
WHERE test_user.id = :id_1
In [71]:
query = User.__table__.insert().values(id = 1, name = 'wxnacy', score = 1)
print(query)

INSERT INTO test_user (id, name, score) VALUES (:id, :name, :score)
In [73]:
query = User.__table__.insert(values=dict(id = 1, name = 'wxnacy', score = 1))
print(query)

INSERT INTO test_user (id, name, score) VALUES (:id, :name, :score)
In [80]:
# 将多个插入语句合并为一个，并执行
items = [{"name": "wxnacy"}, {"name": "wen"}]
session.execute(User.__table__.insert(), items)
Out[80]:
<sqlalchemy.engine.result.ResultProxy at 0x100482c90>
In [89]:
query = User.__table__.update().where(User.id == 1).values(name = 'wxnacy')
print(query)

UPDATE test_user SET name=:name WHERE test_user.id = :id_1
In [93]:
query = User.__table__.update(whereclause=User.id == 1, values=dict(name = 'wxnacy'))
print(query)

UPDATE test_user SET name=:name WHERE test_user.id = :id_1
In [95]:
query = User.__table__.delete()
print(query)

DELETE FROM test_user
In [97]:
query = User.__table__.delete().where(User.id == 1)
print(query)

DELETE FROM test_user WHERE test_user.id = :id_1
In [98]:
query = User.__table__.delete(whereclause=User.id == 1)
print(query)

DELETE FROM test_user WHERE test_user.id = :id_1

导入模块用法

In [9]:
from sqlalchemy import select
query = select([User])
print(query)

SELECT test_user.id, test_user.name, test_user.score 
FROM test_user
In [25]:
query = select([User.id, User.name])
print(query)

SELECT test_user.id, test_user.name 
FROM test_user

更多查询样例

直接传参

In [22]:
query = select([User.id, User.name], User.id == 1)
print(query)

SELECT test_user.id, test_user.name 
FROM test_user 
WHERE test_user.id = :id_1
In [30]:
from sqlalchemy import and_
query = select([User.id, User.name], and_(User.id == 1, User.name == 'wxnacy'))
print(query)

SELECT test_user.id, test_user.name 
FROM test_user 
WHERE test_user.id = :id_1 AND test_user.name = :name_1
In [40]:
from sqlalchemy import text
query = select([User.id, User.name], text("id = 1 and name = 'wxnacy'"))
print(query)

SELECT test_user.id, test_user.name 
FROM test_user 
WHERE id = 1 and name = 'wxnacy'
In [41]:
from sqlalchemy import text
query = select([User.id, User.name], text("id = :id and name = :name")).params(id = 1, name= 'wxnacy')
print(query)

SELECT test_user.id, test_user.name 
FROM test_user 
WHERE id = :id and name = :name

使用 where

In [26]:
query = select([User.id, User.name]).where(User.id == 1)
print(query)

SELECT test_user.id, test_user.name 
FROM test_user 
WHERE test_user.id = :id_1
In [28]:
from sqlalchemy import and_
query = select([User.id, User.name]).where(and_(User.id == 1, User.name == 'wxnacy'))
print(query)

SELECT test_user.id, test_user.name 
FROM test_user 
WHERE test_user.id = :id_1 AND test_user.name = :name_1
In [35]:
from sqlalchemy import text
query = select([User.id, User.name]).where(text("id = 1 and name = 'wxnacy'"))
print(query)

SELECT test_user.id, test_user.name 
FROM test_user 
WHERE id = 1 and name = 'wxnacy'
In [45]:
from sqlalchemy import text
query = select([User.id, User.name]).where(text("id = :id and name = :name")).params(id = 1, name= 'wxnacy')
print(query)

SELECT test_user.id, test_user.name 
FROM test_user 
WHERE id = :id and name = :name

其他

In [44]:
query = select([User.id, User.name]).limit(10)
print(query)

SELECT test_user.id, test_user.name 
FROM test_user
 LIMIT :param_1
In [47]:
query = select([User.id, User.name]).order_by(User.id)
print(query)

SELECT test_user.id, test_user.name 
FROM test_user ORDER BY test_user.id
In [50]:
query = select([User.id, User.name]).order_by(User.id.desc())
print(query)

SELECT test_user.id, test_user.name 
FROM test_user ORDER BY test_user.id DESC
In [52]:
query = select([User.id, User.name]).group_by(User.name)
print(query)

SELECT test_user.id, test_user.name 
FROM test_user GROUP BY test_user.name

更多用法详见文档