标签:type 传递 ring pool make ati types threading 用法
‘‘‘ # &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&& # SQLAlchemy ORM框架(需要依赖数据库API语言) # 架构图 # SQLalchemy orm # SQLalchemy core # --> schema/types 架构/类型 # --> sql expression language SQL表达式语言 # --> engine(框架引擎) # --> connection pooling (数据库连接池) # --> diaiect (选择连接数据路的DB api种类) # DBAPI(pymysql,oracle,sqlite) 目标: 类/对箱操作 -> SQL -> pymsql、mysqldb -> 在去数据中执行 连接数据库: MYSQL - python mysql+mysqldb://<user>:<password>@<host>:<port>/<dbname> pymysql mysql+pymsql://<username>:<password>@<host>/<dbname> 示例: "mysql+pymysql://root:123456@127.0.0.1:3306/t1?charset=utf8" cx_Oracle oracle+cx_oracle://user:pwd@host:port/dbname 基本使用 import sqlalchemy from sqlalchemy import create_engine engine = create_engine import sqlalchemy from sqlalchemy import create_engine import pymysql import threading # 基本连接操作:(不推荐这种用法!) engine = create_engine( "mysql+pymysql://root:123456@127.0.0.1:3306/anec?charset=utf8", max_overflow=2, pool_size=5, pool_timeout=30, pool_recycle=-1, ) def task(arg): conn = engine.raw_connection() cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) cursor.execute(‘select id,name from student;‘) result = cursor.fetchall() print(result) cursor.close() conn.close() for i in range(10): t = threading.Thread(target=task,args=(i,)) t.start() #------------------------------ # 基本连接操作:(推荐这种用法!) # import sqlalchemy # from sqlalchemy import create_engine # from sqlalchemy.orm import scoped_session # from sqlalchemy.orm import sessionmaker # # engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/anec?charset=utf8", pool_size = 5) # Session = sessionmaker(bind=engine) # # # 这种方式的话,就是多线程的,scoped_session 传递Session值至类,封装了treading.local() ,从而达到多线程 # sess = scoped_session(Session) # obj = modeles.表名(name=‘axlex‘,email=‘123@qq.com‘) # sess.add(obj) # sess.commit() #------------------------------ # 创建数据表 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base # from sqlalchemy import Column,Integer,String...... from sqlalchemy import * Base = declarative_base() class Info(Base): __tablename__=‘info‘ id = Column(Integer,primary_key=True) name = Column(String(64),index=True,nullable=False) def init_db(): """ 用类 创建数据表 :return: """ engine = create_engine( "mysql+pymysql://root:123456@127.0.0.1:3306/anec?charset=utf8", max_overflow=2, pool_size=5, pool_timeout=30, pool_recycle=-1, ) Base.metadata.create_all(engine) def drop_db(): """ 用类 创建数据表 :return: """ engine = create_engine( "mysql+pymysql://root:123456@127.0.0.1:3306/anec?charset=utf8", max_overflow=2, pool_size=5, pool_timeout=30, pool_recycle=-1, ) Base.metadata.drop_all(engine) # sqlalchemy默认不能修改表!!!!!!! # init_db() # ---------------------------------- # 操作数据 from sqlalchemy.orm import sessionmaker engine = create_engine( "mysql+pymysql://root:123456@127.0.0.1:3306/anec?charset=utf8", max_overflow=2, pool_size=5, pool_timeout=30, pool_recycle=-1, ) conn = sessionmaker(bind=engine) session = conn() # 添加单条数据 # obj1 = Info(name=‘anec‘) # session.add(obj1) # 添加多条数据 # data = [ # Info(name=‘adasda‘), # Info(name=‘aaaaa‘), # Info(name=‘bbbbb‘), # ] # session.add_all(data) # 查询数据 # 查询所有数据 result = session.query(表名).all() for i in result; print(i.id) print(i.name) # 条件查询数据 result = session.query(表名).filter(表名.字段 > 2) # 删除数据 result = session.query(表名).filter(表名.字段 > 2).delete() # 修改数据 result = session.query(表名).filter(表名.字段 = 2).update({‘字段‘:"新数值"}) result = session.query(表名).filter(表名.字段 > 2).update({‘字段‘:"字符字段"+字符},synchronize_session=False) result = session.query(表名).filter(表名.字段 > 2).update({‘字段‘:"数值字段"+数值},synchronize_session=evaluate) session.commit() session.close() ‘‘‘
标签:type 传递 ring pool make ati types threading 用法
原文地址:https://www.cnblogs.com/Anec/p/10469923.html