自定义 Mysql 类 与 自定义 异常类
2024-10-16 16:51:04
import MySQLdb
class MyExcept(Exception): ''' 常见做法定义异常基类,然后在派生不同类型的异常 '''
def __init__(self, *args): self.args = args
class DropDataaseError(MyExcept): def __init__(self): self.args = ('删除数据库错误!',) self.message = '删除数据库错误!' self.code = 100
class DropTableError(MyExcept): def __init__(self): self.args = ('删除表错误!',) self.message = '删除表错误！' self.code = 200
class CreateDatabaseError(MyExcept): def __init__(self): self.args = ('不能创建数据库',) self.message = '不能创建数据库' self.code = 300
class OperatorError(MyExcept): ''' 操作错误,一般是要做的事情和实际功能不匹配 '''
def __init__(self, message): self.args = (message,) self.message = message self.code = 400
class FileIsExistsError(MyExcept): def __init__(self, message): self.args = (message,) self.message = message self.code = 500
##raise OperatorError('错误消息') # 这里突然返现 raise引发的异常将中断程序import jsonimport os
class Mysql: conn = None host = None port = None user = None passwd = None charset = None mysql_name = None cursor = None
@classmethod def create_database(cls, sql): if 'create database' not in sql: raise OperatorError('这是创建数据库') cls.query(sql) cls.use_database()
@classmethod def use_database(cls): if cls.mysql_name != None: cls.query('use %s ;' % cls.mysql_name) else: raise OperatorError('切换数据库失败')
@classmethod def config(cls, mode=False, host='localhost', port=3306, mysql_name='test', user='root', passwd='', charset='utf8', path='config.json'): ''' 连接数据库配置文件 :param mode: mode false 默认为自动传值，true 为从json配置文件里面获取数据 :param host:ip 默认为空localhost :param port: 端口 默认为3306 :param mysql_name: 数据库名称 必须指定 :param user: 用户名称，默认值为root :param passwd: 默认密码为空 :param charset: 默认字符编码为utf8 :param cursorclass:Faslse 不指定MySQLdb.cursors.DictCursor，True指定为MySQLdb.cursors.DictCursor :return: NUll '''
if not mode:
if {'', ' ', ' ', None} & {host, port, user, charset, mysql_name}: raise OperatorError( "host,port,user,charset可以不传值，但是不能传空值!,也不可为空格")
Mysql.host = host Mysql.port = port Mysql.user = user Mysql.passwd = passwd Mysql.charset = charset Mysql.mysql_name = mysql_name
if mode: if not os.path.exists(path): raise FileIsExistsError('文件不存在')
with open(path, 'rt', encoding='utf8') as f: config_json_dic = json.load(f)
Mysql.host = config_json_dic['host'] Mysql.port = config_json_dic['port'] Mysql.user = config_json_dic['user'] Mysql.passwd = config_json_dic['passwd'] Mysql.charset = config_json_dic['charset']
try: Mysql.conn = MySQLdb.connect(host=Mysql.host, port=Mysql.port, user=Mysql.user, passwd=Mysql.passwd, db=Mysql.mysql_name, charset=Mysql.charset) except Exception as e: Mysql.conn.close() print(e) else: Mysql.get_cursor() # cls.use_database()
@classmethod def get_cursor(cls, mode=False): # -------------------------- ''' 获取操作的游标 :return: Null ''' if not mode: Mysql.cursor = Mysql.conn.cursor() else: Mysql.cursor = Mysql.conn.cursor( cursorclass=MySQLdb.cursors.DictCursor)
@classmethod def having_sql(cls, sql): ''' sql 过滤 :param sql:需要过滤的sql :return:过滤后的sql ''' # 没有写好 return True
@classmethod def query(cls, sql): ''' 底层 sql 执行语句 :param sql: sql语句 :return: ''' # if 'drop table' in sql: # raise ValueError('不允许删除表!') # # if 'drop database' in sql: # raise ValueError('不允许删除数据库')
if {'', ' ', ' ', None} & {Mysql.host, Mysql.port, Mysql.user, Mysql.charset, Mysql.mysql_name}: raise ValueError("host,port,user,charset可以不传值，但是不能传空值!")
if not cls.having_sql(sql): raise ValueError('含有非法字符')
try: Mysql.cursor.execute(sql) Mysql.conn.commit() # 提交 except: Mysql.conn.rollback() # 回滚操作 return False else: return True
@classmethod def update(cls, sql): if not ('update' in sql): raise OperatorError('更新表记录！')
return Mysql.query(sql)
@classmethod def truncate(cls, sql): if not ('truncate' in sql): raise OperatorError('清空表记录操作')
return Mysql.query(sql)
@classmethod def delete(cls, sql): if not ('delete' in sql): raise OperatorError('删除表记录！')
return Mysql.query(sql)
@classmethod def create(cls, sql): if not ('create table' in sql): raise OperatorError('创建表！')
if 'create database' in sql: raise CreateDatabaseError()
return Mysql.query(sql)
@classmethod def insert(cls, sql): if not ('insert into' in sql): raise OperatorError('插入记录！')
return Mysql.query(sql)
@classmethod def insert_many(cls, sql, res): ''' 批量插入 :param sql:sql sample:insert into talbe_name values(0,%s,%s,%s,%s,%s) :param res:res sample:((title,singer,imgurl,url,alpha),(title2,singer2,imgurl2,url2,alpha2)) :return line: why? ''' if not ('insert into' in sql): raise OperatorError('操作错误，这是插入') if not len(res) or len(res) < 5: raise ValueError('插入的值不能为空')
line = Mysql.cursor.executemany(sql, res) return line
@classmethod def get_line(cls, sql): ''' 获取单行的数据 :return: ''' if not ('select' in sql): raise OperatorError('执行查询')
cls.query(sql)
try: return True, Mysql.cursor.fetchone(); except: return False, '没有数据'
@classmethod def get_lines(cls, sql): ''' 返回多行数据 :return: ''' if not ('select' in sql): raise OperatorError('执行查询')
cls.query(sql)
try: return True, Mysql.cursor.fetchall(); except Exception as e: print(e) return False, '没有数据'
@classmethod def get_fetchmany(cls, sql, size=1): ''' 获取指定数量数据 :param size: 接收size条返回结果行.如果size的值大于返回的结果行的数量,则会返回cursor.arraysize条数 :return:tuple ''' if not isinstance(size, int): raise TypeError('类型错误')
if size <= 0: return None;
if not ('select' in sql): raise OperatorError('执行查询')
cls.query(sql)
return Mysql.cursor.fechmany(size)
@classmethod def close(cls): ''' 关闭cursor 和db 连接 :return: ''' Mysql.cursor.close() Mysql.conn.close()
# # @classmethod # def __del__(): # Mysql.close()
最新文章
- ajax 同步和异步
- HDU 2509 Be the Winner nim博弈变形
- Adapter 代码中启动报错
- L440 无线网卡：由于该设备有问题，Windows 已将其停止(代码 43)
- Windows 2003 + IIS6.0 相关 401.1 或 401.2 等问题解决
- boost:进程管理
- c++异常安全和copy and swap策略
- nginx 3.nginx+fastcgi
- NYOJ 14 场地安排（它可以被视为一个经典问题）
- IOS NSArray 倒序
- Windows Server 2012的配置与部署
- 12款Linux系统恢复工具
- 关于mysql增删改查的基本操作
- bzoj 4289: PA2012 Tax
- wpf中静态资源和动态资源的区别
- 一例完全理解vue 2.0 的slots 和 functional render
- Dynamics CRM2013/2015 禁止欢迎界面（Disable the Welcome Screen）
- express+handlebars 快速搭建网站前后台
- python接口测试-认识GET请求
- javaScript:压缩图片并上传