首页 > 代码库 > python SQLAlchemy

python SQLAlchemy

这里我们记录几个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:tiger@localhost/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

python SQLAlchemy