关于 sqlalchemy 快速入门总结

熊孩纸 阅读:191 2021-06-09 13:42:50 评论:0

主要对于粗略看了一下文档,但是又不是特别清楚的人提供一些信息。

简单认识

创建连接对象
from sqlalchemy import create_engine 
from sqlalchemy.orm import sessionmaker 
mysql_conf = {
    
    'host': '127.0.0.1', 
    'user': 'root', 
    'passwd': 'root', 
    'db': 'diyidan_link_service', 
    'port': '3306', 
    # mysql中的utf8不是真正的utf8,建议直接使用utf8mb4 
    'charset': 'utf8mb4', 
} 
engine = create_engine( 
    'mysql+pymysql://%(user)s:%(passwd)s@%(host)s:%(port)s/%(db)s?charset=%(charset)s' % mysql_conf, 
    echo=True,				# 打印sql语句(正式服不要加) 
    max_overflow=0,  		# 超过连接池大小外最多创建的连接 
    pool_size=5,  			# 连接池大小 
    pool_timeout=30,  		# 池中没有线程最多等待的时间,否则报错 
    pool_recycle=-1  		# 多久之后对线程池中的线程进行一次连接的回收(重置) 
) 
Session = sessionmaker(bind=engine) 
# 大项目最好自己定义一个session工厂方法。 
session = Session() 
建表
import sqlalchemy 
from sqlalchemy import Column 
from sqlalchemy import String,TIMESTAMP,Text,BIGINT 
from sqlalchemy.ext.declarative import declarative_base 
 
# 相当于django orm中的Model类(可以利用declarative_base()定义公共字段) 
BaseModel = declarative_base() 
 
class TestModel(BaseModel): 
	__tablename__ = 'test' 
    __table_args__ = {
   'mysql_charset': 'utf8', 'mysql_engine': 'InnoDB'} 
	 
	# 主键 
	test_id = Column(BIGINT(20, unsigned=True), primary_key=True) 
	title = Column(String(50), nullable=True) 
	content = Column(Text, nullable=True) 
	# 创建时自动添加 
	create_time = Column(TIMESTAMP, server_default=sqlalchemy.sql.expression.text('CURRENT_TIMESTAMP()')) 
	# 修改时自动更新 
    update_time = Column(TIMESTAMP, server_default=sqlalchemy.sql.expression.text('CURRENT_TIMESTAMP() on update CURRENT_TIMESTAMP()')) 
生成表
BaseModel.metadata.create_all(engine) 
增删改查
############### 增 
tx = TestModel(title="测试", content="测试insert") 
session.add(tx) 
session.commit() 
 
tx1 = TestModel(title="测试1", content="测试insert") 
tx2 = TestModel(title="测试2", content="测试insert") 
tx3 = TestModel(title="测试3", content="测试insert") 
session.add_all([tx1,tx2,tx3]) 
session.commit() 
 
############### 查 
tx_query_obj = session.query(		# 相当于 select 
	TestModel    # 查出所有字段 
	# TestModel.title	 # 查出指定字段 
	# TestModel.title.label('the_title')	# 查出指定字段并重命名 
	).filter(						# 相当于where 过滤器 
	a == b 
	).filter_by(					# 相当于where 等值过滤器 
	a = b 
	) 
# 此时 tx_query_obj 并未执行sql语句。只有在使用数据时才会执行。这种惰性加载和django orm是类似的 
tx_query_all = tx_query_obj.all()		# 全部查询结果			list 
tx_query_first = tx_query_obj.first()	# 第一条查询结果			obj 
tx_query_one = tx_query_obj.one()		# 获取目标结果			obj 不存在或多行报错 
tx_query_one_or_one = tx_query_obj.one_or_one()		# 获取目标结果 与one 类似,不存在返回None 
tx_query_scalar = tx_query_obj.scalar()		#调用one,成功则返回第一列 
 
############### 改 
tx3.title = "测试4" 
session.commit() 
 
############### 删 
session.delete(tx2) 
session.commit() 

你需要了解 filter 与 filter_by 的区别

有意思的操作

按时间分组
from sqlalchemy import func 
# 按创建时间的天分组 同理 func.month func.year func.hour ... 
query.group_by(func.day(TestModel.create_time)) 

注意点

filter 与 filter_by 的区别

filter 过滤器
   只接受一个条件,key 必须是 模型的属性对象 如 TestModel.title
   多个条件可以使用多个filter,也可以使用一些内置函数连接多条件
   接收条件可以为 大于(>) ,小于(<), 等于(==),不等于(!=)等
   逻辑运算支持 in_, is_, and_ or_ 等
   如:query.filter(TestModel.title.in_([‘测试’, ‘测试1’, ‘测试2’]))
   加~符号表示not,query.filter(TestModel.title.in_([‘测试’, ‘测试1’, ‘测试2’]))
filter_by 等值过滤器:query.filter(~TestModel.title.in_([‘测试’, ‘测试1’, ‘测试2’]))
   只接收等值条件,可以接收 多个 key=value 形式等值参数。
   key为字段字符串即可,即可以使用 **kw 形式传参。

用到一点就学一点然后更新一点。。。


标签:Python
声明

1.本站遵循行业规范,任何转载的稿件都会明确标注作者和来源;2.本站的原创文章,请转载时务必注明文章作者和来源,不尊重原创的行为我们将追究责任;3.作者投稿可能会经我们编辑修改或补充。

发表评论
搜索
KIKK导航

KIKK导航

关注我们