python 模块 SQLalchemy

Posted 2021-02-11 Anec

 

‘‘‘
# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&
# 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:[email protected]:3306/t1?charset=utf8"
   
    cx_Oracle
    oracle+cx_oracle://user:[email protected]: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:[email protected]: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:[email protected]: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=‘[email protected]‘)
# 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:[email protected]: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:[email protected]: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:[email protected]: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()


‘‘‘