python 模块 SQLalchemy
时间：2021-07-01 10:21:17
帮助过：21人阅读
# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&
# 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()
‘‘‘
python 模块 SQLalchemy
标签：type 传递 ring pool make ati types threading 用法
