from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
class MySQLClient(object):
"""MySQL数据库客户端"""
def __init__(self, host, port, dbname, username, password, **setting):
self.engine = create_engine("mysql+pymysql://{username}:{password}@{host}:{port}/{dbname}?charset={encoding}"
.format(host=host, port=port, dbname=dbname, username=username, password=password,
encoding=setting.pop('encoding', 'utf8')),
**setting)
self._session = sessionmaker(bind=self.engine)
def query(self, statement):
"""查询所有"""
session = self._session()
try:
return session.query(statement).all()
finally:
session.close()
def query_all_app_job(self):
"""
查询所有应用任务
:return:
"""
session = self._session()
try:
return session.query(TVisualAppJob).all()
finally:
session.close()
def query_all_host_job(self):
"""
查询所有应用任务
:return:
"""
session = self._session()
try:
return session.query(TVisualHostJob).all()
finally:
session.close()
def query_all_app(self):
"""
查询所有应用
:return:
"""
session = self._session()
try:
return session.query(TCollectApp).all()
finally:
session.close()
def query_all_host(self):
"""
查询所有主机
:return:
"""
session = self._session()
try:
return session.query(TCollectHost).all()
finally:
session.close()
def query_app_contact(self, app_job_id):
"""
根据主机任务查询联系人
:param host_job_id:
:return:
"""
session = self._session()
try:
return session.query(TVisualAppJob).get(app_job_id).contacts
finally:
session.close()
def query_host_contact(self, host_job_id):
"""
根据主机任务查询联系人
:param host_job_id:
:return:
"""
session = self._session()
try:
return session.query(TVisualHostJob).get(host_job_id).contacts
finally:
session.close()
def find_user(self, username, password):
"""判断用户在数据库是否存在"""
session = self._session()
try:
return session.query(TVisualUser).filter_by(username=username, password=password).one_or_none()
finally:
session.close()
def update_user(self, username, login_time, login_ip):
"""更新用户登录日期和ip"""
session = self._session()
try:
session.query(TVisualUser).filter_by(username=username).update(
{'last_login_at': login_time, 'last_login_ip': login_ip})
session.commit()
except Exception as e:
session.rollback()
raise e
finally:
session.close()
def close(self):
pass
SqlAlchemy笔记(基于Tornado)
猜你喜欢
转载自blog.csdn.net/layswift/article/details/80923752
今日推荐
周排行