马上就要开始写仿优酷系统了，在这里提前学习一下ORM。
一开始去百度搜了一下相关资料，里面的ORM框架要自己写，所以花了半天的时间学习了sqlalchemy:
SQLAlchemy
SQLAlchemy 比 Django 自带的 ORM 好在哪里?
python sqlalchemy 和django orm 对比
不管是说Django的ORM好，还是说Sqlalchemy好，，我认为框架并不是重点，其核心在于通过学习一门ORM框架，达到能熟练的从数据库取出数据保存数据（也就是数据持久化），这才是重点。
Aiomysql:
同样道理，掌握一门异步库也能让自己更加首席线程的操作：
1.为什么要使用异步访问数据库，有什么优缺点？
在多用户高并发的情况下，若使用同步查询，则等待时间过长，代价很高；
使用了异步能提高效率，但在同时增加了处理难度；而且还会有“读脏数据的风险”：
A和B要买票，同时A和B都发现只剩一张票了，这个时候两人同时点击购买，则会出现“读脏数据”的情况
2.普通的线程池和Aiomysql有什么区别?
首先准确的来说，Aiomysql主要使用的是协程，其次才是多线程，多线程建立连接池，每个连接使用协程处理多并发的情况，
所以，他们的区别应该是在，普通的线程池的每一个线程没有使用协程，而aiomysql线程池里的线程使用了协程
正式开始主题：
1.Aiomysql创建连接
class DBcontroller: __engine=None __isinstance = False def __new__(cls, *args, **kwargs): if cls.__isinstance: # 如果被实例化了 return cls.__isinstance # 返回实例化对象 print('connecting to database...') asyncio.get_event_loop().run_until_complete(DBcontroller.connect()) cls.__isinstance = object.__new__(cls) # 否则实例化 return cls.__isinstance # 返回实例化的对象 @staticmethod async def connect(): try: __engine = await create_engine(user='root', db='youku', host='127.0.0.1', password='root', minsize=1, maxsize=10, autocommit=True) if __engine: DBcontroller.__engine = __engine DBcontroller.connectStatue =True print('connect to mysql success!') else: raise ("connect to mysql error ") except: print('connect error.', exc_info=True)
2.创建Model
class User(Base,Model):
__tablename__ = 'user'
id = Column(INTEGER(11), primary_key=True)
name = Column(String(255))
password = Column(String(255))
is_locked = Column(INTEGER(11), nullable=False, server_default=text("'0'"))
is_vip = Column(INTEGER(11), nullable=False, server_default=text("'0'"))
user_type = Column(String(255))
register_time = Column(String(255))
3.创建sqlalchemy表
from sqlalchemy import Column, Integer, String, MetaData,Table metadata = sa.MetaData() user = Table('user', MetaData(), Column('id', Integer, primary_key=True), Column('name', String(50)), Column('is_locked', Integer,nullable=False,default=0), Column('is_vip', Integer,nullable=False,default=0), Column('user_type', String(12)), Column('register_time', String(12)) ) ()
4.实现查询/更新/插入功能
class DBcontroller:
__engine=None
__isinstance = False
def __new__(cls, *args, **kwargs):
if cls.__isinstance: # 如果被实例化了
return cls.__isinstance # 返回实例化对象
print('connecting to database...')
asyncio.get_event_loop().run_until_complete(DBcontroller.connect())
cls.__isinstance = object.__new__(cls) # 否则实例化
return cls.__isinstance # 返回实例化的对象
@staticmethod
async def connect():
try:
__engine = await create_engine(user='root',
db='youku',
host='127.0.0.1',
password='root',
minsize=1,
maxsize=10,
autocommit=True)
if __engine:
DBcontroller.__engine = __engine
DBcontroller.connectStatue =True
print('connect to mysql success!')
else:
raise ("connect to mysql error ")
except:
print('connect error.', exc_info=True)
def selectTable(self,table):
async def select():
dbstb={
'user':user
}
conn = await DBcontroller.__engine.acquire()
try:
result = await conn.execute(dbstb[table].select())
res = await result.fetchall()
for row in res:
print(row)
except Exception as e :
print(e)
finally:
# DBcontroller.__engine.release(conn)
pass
asyncio.get_event_loop().run_until_complete(select())
def executeTable(self,table,attr):
async def execute():
dbstb={
'user':user
}
conn = await DBcontroller.__engine.acquire()
try:
await conn.execute(dbstb[table].insert().values(attr))
except Exception as e:
if e.args[0]== 1062:
print('主键已存在')
elif e.args[0]==1366:
if attr['id']=='':
print('未设置主键')
elif attr['is_vip']=='':
print('未设置vip类型')
elif attr['is_locked']=='':
print('未设置上锁类型')
else:print('未知错误，请联系管理员')
else:
print(f'connect failed:{e}')
finally:
DBcontroller.__engine.release(conn)
asyncio.get_event_loop().run_until_complete(execute())
def updateTable(self,table,id,attr):
async def update():
dbstb={
'user':user
}
conn = await DBcontroller.__engine.acquire()
try:
await conn.execute(user.update().where(dbstb[table].c.id==id).values(attr))
except Exception as e:
if e.args[0]== 1062:
print('主键已存在')
elif e.args[0]==1366:
if attr['id']=='':
print('未设置主键')
elif attr['is_vip']=='':
print('未设置vip类型')
elif attr['is_locked']=='':
print('未设置上锁类型')
else:print('未知错误，请联系管理员')
print(e)
else:
print(f'connect failed:{e}')
finally:
DBcontroller.__engine.release(conn)
asyncio.get_event_loop().run_until_complete(update())
结果
