2017-05-27

Python

3 分钟读完 (大约 494 个字)

sqlalchemy使用小结

本文记录了sqlalchemy基础常用的一些操作

创建模型

from datetime import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DATETIME, PrimaryKeyConstraint
Base = declarative_base()

class User(Base):
    __tablename__ = 'user' #表名
    __table_args__ = (
        PrimaryKeyConstraint('user_id', name='PK_USER_ID'), # user_id为主键名称
    )
    
    user_id = Column(Integer)
    user_name = Cloumn(String(50), nullable=False)
    user_pwd = Column(String(50), nullable=False)
    create_time = Column(DATETIME, default=datetime.now)

创建连接

from sqlalchemy import create_engine
frmo sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URI = \
    "mysql+mysqldb://{name:s}:{pw:s}@{addr:s}:{port}/{db:s}?charset={charset:s}"\
    .format(
        name='test',
        pw='123456',
        addr='127.0.0.1',
        port=3306,
        db='test_db',
        charset='utf8',
    )

engine = create_engine(SQLALCHEMY_DATABASE_URI,
    echo=True, #是否显示执行的SQL语,
    pool_size=0, #连接池大小，0表示无限制,
    max_overflow=-1, #连接池额可增加上限,
    pool_recycle=120 #连接池回收连接时间
)

db_session_mk = sessionmaker(bind=engine)

db_sesion = db_session_mk()

查询

query = db_session.query(User)

# 获取一个
query.first()

# 获取所有
query.all()

# 指定下标
query[10:20]

带条件的查询

and

query = db_session.query(User)

# filter参数为and逻辑
query.filter(
    User.user_name == 'root',
    User.user_pwd == '123456'
).first()

# 指定
from sqlalchemy import and_
query.filter(and_(
    User.user_name == 'root',
    User.user_pwd == '123456'
))

from sqlalchemy import or_

query.filter(or_(
    User.user_name == 'root',
    User.user_name == 'admin'
))

查询条件为数组

1	query.filter(User.user_name.in_(['root','admin'])

模糊查询

1	query.filter(User.user_name.like('root%'))

指定查询字段

1	db_session.query(User.user_id, User.user_name).all()

排序

1	from sqlalchemy import desc, asc
2
3	query.order_by(desc(User.user_id)).all()
4	query.order_by(asc(User.user_id)).all()

删除

db_session.query(User).filter(
    User.user_id == 1
).delete(synchronize_session=False)

db_session.commit()

修改

#第一种
user = db_session.query(User).filter(User.user_id==1).first()
user.user_name = 'admin'
db_session.commit()

#第二种
db_session.query(User).filter(User.user_id==1)\
    .update({
        User.user_name:'admin'
    }, synchronize_session=False)
db_session.commit()

添加

1	user = User()
2	user.user_name = 'root'
3	db_session.add(user)
4	db_session.commit()

批量添加

1	root = User()
2	admin = User()
3	db_session.bulk_save_objects([root, admin])
4	db_session.commit()

事务

sqlalchmey默认是开启事务的

root = User()
root.user_name = 'root'

admin = User()
admin.user_name = 'admin'

db_session.add(root)
db_session.add(admin)

try:
    db_session.commit()
except:
    db_session.rollback()
finally:
    db_session.close()

本文标题：sqlalchemy使用小结
本文作者：W
本文链接：http://skrnet.cn/articles/sqlalchemy_usage_reference.html
发布时间：2017-05-27
版权声明：本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明出处！

# sqlalchemy

sqlalchemy使用小结

创建模型

创建连接

查询

带条件的查询

指定查询字段

排序

删除

修改

添加

批量添加

事务

评论

链接

分类

标签云

最新文章

归档

标签

最新文章

归档

标签

Your browser is out-of-date!

1	from datetime import datetime
2	from sqlalchemy.ext.declarative import declarative_base
3	from sqlalchemy import Column, Integer, String, DATETIME, PrimaryKeyConstraint
4	Base = declarative_base()
5
6	class User(Base):
7	__tablename__ = 'user' #表名
8	__table_args__ = (
9	PrimaryKeyConstraint('user_id', name='PK_USER_ID'), # user_id为主键名称
10	)
11
12	user_id = Column(Integer)
13	user_name = Cloumn(String(50), nullable=False)
14	user_pwd = Column(String(50), nullable=False)
15	create_time = Column(DATETIME, default=datetime.now)

1	from sqlalchemy import create_engine
2	frmo sqlalchemy.orm import sessionmaker
3
4	SQLALCHEMY_DATABASE_URI = \
5	"mysql+mysqldb://{name:s}:{pw:s}@{addr:s}:{port}/{db:s}?charset={charset:s}"\
6	.format(
7	name='test',
8	pw='123456',
9	addr='127.0.0.1',
10	port=3306,
11	db='test_db',
12	charset='utf8',
13	)
14
15	engine = create_engine(SQLALCHEMY_DATABASE_URI,
16	echo=True, #是否显示执行的SQL语,
17	pool_size=0, #连接池大小，0表示无限制,
18	max_overflow=-1, #连接池额可增加上限,
19	pool_recycle=120 #连接池回收连接时间
20	)
21
22	db_session_mk = sessionmaker(bind=engine)
23
24	db_sesion = db_session_mk()

1	query = db_session.query(User)
2
3	# 获取一个
4	query.first()
5
6	# 获取所有
7	query.all()
8
9	# 指定下标
10	query[10:20]

1	query = db_session.query(User)
2
3	# filter参数为and逻辑
4	query.filter(
5	User.user_name == 'root',
6	User.user_pwd == '123456'
7	).first()
8
9	# 指定
10	from sqlalchemy import and_
11	query.filter(and_(
12	User.user_name == 'root',
13	User.user_pwd == '123456'
14	))

1	from sqlalchemy import or_
2
3	query.filter(or_(
4	User.user_name == 'root',
5	User.user_name == 'admin'
6	))

1	db_session.query(User).filter(
2	User.user_id == 1
3	).delete(synchronize_session=False)
4
5	db_session.commit()

1	#第一种
2	user = db_session.query(User).filter(User.user_id==1).first()
3	user.user_name = 'admin'
4	db_session.commit()
5
6	#第二种
7	db_session.query(User).filter(User.user_id==1)\
8	.update({
9	User.user_name:'admin'
10	}, synchronize_session=False)
11	db_session.commit()

1	root = User()
2	root.user_name = 'root'
3
4	admin = User()
5	admin.user_name = 'admin'
6
7	db_session.add(root)
8	db_session.add(admin)
9
10	try:
11	db_session.commit()
12	except:
13	db_session.rollback()
14	finally:
15	db_session.close()