python SQLAlchemy

Posted 2020-08-16

这里我们记录几个python SQLAlchemy的使用例子：

如何对一个字段进行自增操作

user = session.query(User).with_lockmode(‘update‘).get(1)
user.age += 1
session.commit()

对多条数据进行删除操作：

session.query(User).filter(or_(User.id == 1, User.id == 2, User.id == 3)).delete()

各种连接表，并进行多表操作：

from sqlalchemy import distinct
from sqlalchemy.orm import aliased

 

Friend = aliased(User, name=‘Friend‘)

print session.query(User.id).join(Friendship, User.id == Friendship.user_id1).all() # 所有有朋友的用户
print session.query(distinct(User.id)).join(Friendship, User.id == Friendship.user_id1).all() # 所有有朋友的用户（去掉重复的）
print session.query(User.id).join(Friendship, User.id == Friendship.user_id1).distinct().all() # 同上
print session.query(Friendship.user_id2).join(User, User.id == Friendship.user_id1).order_by(Friendship.user_id2).distinct().all() # 所有被别人当成朋友的用户
print session.query(Friendship.user_id2).select_from(User).join(Friendship, User.id == Friendship.user_id1).order_by(Friendship.user_id2).distinct().all() # 同上，join 的方向相反，但因为不是 STRAIGHT_JOIN，所以 MySQL 可以自己选择顺序
print session.query(User.id, Friendship.user_id2).join(Friendship, User.id == Friendship.user_id1).all() # 用户及其朋友
print session.query(User.id, Friendship.user_id2).join(Friendship, User.id == Friendship.user_id1).filter(User.id < 10).all() # id 小于 10 的用户及其朋友

 

外键的建立

from random import randint
from sqlalchemy import ForeignKey

 

class User(BaseModel):
    __tablename__ = ‘user‘

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

class Friendship(BaseModel):
    __tablename__ = ‘friendship‘

    id = Column(Integer, primary_key=True)
    user_id1 = Column(Integer, ForeignKey(‘user.id‘))
    user_id2 = Column(Integer, ForeignKey(‘user.id‘))

 

何替换一个已有主键的记录？

使用 session.merge() 方法替代 session.add()，其实就是 SELECT + UPDATE：

user = User(id=1, name=‘ooxx‘)
session.merge(user)
session.commit()

 

Table(‘mytable‘, metadata,
      Column(‘data‘, String(32)),
      mysql_engine=‘InnoDB‘,
      mysql_charset=‘utf8‘,
      mysql_key_block_size="1024"
     )

engine = create_engine(
                "mysql://scott:[email protected]/test",
                isolation_level="READ UNCOMMITTED"
            )

To set using per-connection execution options:

connection = engine.connect()
connection = connection.execution_options(
    isolation_level="READ COMMITTED"
)

Valid values for isolation_level include:

• READ COMMITTED
• READ UNCOMMITTED
• REPEATABLE READ
• SERIALIZABLE
• AUTOCOMMIT