python---sqlalchemy(二)

本文主要是介绍python---sqlalchemy(二)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

多对一

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engineBase = declarative_base()class Hobby(Base):__tablename__ = 'hobby'id = Column(Integer, primary_key=True)caption = Column(String(50), default='篮球')class Person(Base):__tablename__ = 'person'nid = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=True)hobby_id = Column(Integer, ForeignKey("hobby.id"))# 与生成表结构无关，仅用于查询方便hobby = relationship("Hobby", backref='pers')def init_db():"""根据类创建数据库表:return:"""engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）)Base.metadata.create_all(engine)if __name__ == '__main__':# drop_db()init_db()engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
####################################多对一添加#######################################
# 添加方式一
# session.add_all([
#     Hobby(caption='乒乓球'),
#     Hobby(caption='羽毛球'),
#     Person(name='张三', hobby_id=1),
#     Person(name='李四', hobby_id=2),
# ])#添加方式二
# person = Person(name='张九', hobby=Hobby(caption='姑娘'))
# session.add(person)# 添加方式三
# hb = Hobby(caption='人妖')
# hb.pers = [Person(name='文飞'), Person(name='博雅')]
# session.add(hb)
# session.commit()# #######################################relationship正向查询#######################################
v = session.query(Person).first()
print(v.name)
print(v.hobby.caption)# #######################################relationship反向查询#######################################
v = session.query(Hobby).first()
print(v.caption)
print(v.pers)session.close()

多对多

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engineBase = declarative_base()class Server2Group(Base):__tablename__ = 'server2group'id = Column(Integer, primary_key=True, autoincrement=True)server_id = Column(Integer, ForeignKey('server.id'))group_id = Column(Integer, ForeignKey('group_.id'))class Group(Base):__tablename__ = 'group_'id = Column(Integer, primary_key=True)name = Column(String(64), unique=True, nullable=False)# 与生成表结构无关，仅用于查询方便servers = relationship('Server', secondary='server2group', backref='groups')class Server(Base):__tablename__ = 'server'id = Column(Integer, primary_key=True, autoincrement=True)hostname = Column(String(64), unique=True, nullable=False)def init_db():"""根据类创建数据库表:return:"""engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）)Base.metadata.create_all(engine)if __name__ == '__main__':# drop_db()init_db()engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
####################################多对一添加######################################## 方式一
# session.add_all([
#     Server(hostname='c1.com'),
#     Server(hostname='c2.com'),
#     Group(name='A组'),
#     Group(name='B组'),
# ])
# session.commit()
#
# s2g = Server2Group(server_id=1, group_id=1)
# session.add(s2g)
# session.commit()# 方式二
# gp = Group(name='C组')
# gp.servers = [Server(hostname='c3.com'), Server(hostname='c4.com')]
# session.add(gp)
# session.commit()# 方式三
# ser = Server(hostname='c6.com')
# ser.groups = [Group(name='F组'),Group(name='G组')]
# session.add(ser)
# session.commit()## # # # # # # # # # # # # # # # #  使用relationship正向查询# # # # # # # # # # # # # # #
v = session.query(Group).first()
print(v.name)
print(v.servers)
for i in v.servers:print(i.hostname)## # # # # # # # # # # # # # # # # # #  使用relationship反向查询# # # # # # # # #
v = session.query(Server).first()
print(v.hostname)
print(v.groups)session.close()

原生语句

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engineBase = declarative_base()class Server2Group(Base):__tablename__ = 'server2group'id = Column(Integer, primary_key=True, autoincrement=True)server_id = Column(Integer, ForeignKey('server.id'))group_id = Column(Integer, ForeignKey('group_.id'))class Group(Base):__tablename__ = 'group_'id = Column(Integer, primary_key=True)name = Column(String(64), unique=True, nullable=False)# 与生成表结构无关，仅用于查询方便servers = relationship('Server', secondary='server2group', backref='groups')class Server(Base):__tablename__ = 'server'id = Column(Integer, primary_key=True, autoincrement=True)hostname = Column(String(64), unique=True, nullable=False)def init_db():"""根据类创建数据库表:return:"""engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）)Base.metadata.create_all(engine)if __name__ == '__main__':# drop_db()init_db()engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)session = Session()# 查询
cursor = session.execute('select * from group_')
result = cursor.fetchall()
print(result)# 添加
cursor = session.execute('insert into group_(name) values(:value)',params={"value":'GG呵呵'})
session.commit()
print(cursor.lastrowid)session.close()

输出

[(3, 'C组'), (1, 'F组'), (4, 'GG'), (2, 'G组')]
5

这篇关于python---sqlalchemy(二)的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---