关于 sqlalchemy 快速入门

爱被打了一巴掌 2022-01-30 05:45 177阅读 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 形式传参。

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