refactor(qd): 🦄 优化代码和 logger 格式和静态类型

.flake8

@@ -0,0 +1,4 @@
+[flake8]
+max-line-length = 120
+ignore = E203, E266, E501, W503
+exclude = .git, __pycache__, venv, dist, build

Pipfile

@@ -37,6 +37,9 @@ tzdata = "*"
 [dev-packages]
 #ddddocr = "*"
 #pycurl = {version = "*", markers="sys_platform != 'win32'"}
+types-croniter = "*"
+types-requests = "*"
+types-python-dateutil = "*"
 
 [requires]
 python_version = "3.11"

backup.py

@@ -4,6 +4,7 @@
 # Author: Binux<[email protected]>
 #         http://binux.me
 # Created on 2014-08-09 11:39:25
+# pylint: disable=broad-exception-raised
 
 import sqlite3
 
@@ -127,30 +128,29 @@ def new(self, userid, maindb):
                 `notepadid` INTEGER NOT NULL ,
                 `content` TEXT NULL
                 );
-                ''' )
+                ''')
 
             # 获取数据库信息
             userid = int(userid)
-            user = maindb.db.user.get(id=userid, fields=('id', 'email', 'email_verified', 'password', 'password_md5', 'userkey', 'nickname', 'role', 'ctime', 'mtime', 'atime', 'cip',
-                                      'mip', 'aip', 'skey', 'barkurl', 'wxpusher', 'noticeflg', 'logtime', 'status', 'notepad', 'diypusher', 'qywx_token', 'tg_token', 'dingding_token', 'qywx_webhook', 'push_batch'))
-            userkey = maindb.db.user.__getuserkey(user['env'])
+            # user = maindb.db.user.get(id=userid, fields=('id', 'email', 'email_verified', 'password', 'password_md5', 'userkey', 'nickname', 'role', 'ctime', 'mtime', 'atime', 'cip',
+            #                           'mip', 'aip', 'skey', 'barkurl', 'wxpusher', 'noticeflg', 'logtime', 'status', 'notepad', 'diypusher', 'qywx_token', 'tg_token', 'dingding_token', 'qywx_webhook', 'push_batch'))
+            # userkey = maindb.db.user.__getuserkey(user['env'])
             tpls = []
             for tpl in maindb.db.tpl.list(fields=('id', 'userid', 'siteurl', 'sitename', 'banner', 'disabled', 'public', 'lock', 'fork', 'har', 'tpl', 'variables', 'interval', 'note', 'success_count', 'failed_count', 'last_success', 'ctime', 'mtime', 'atime', 'tplurl', 'updateable', '_groups', 'init_env'), limit=None):
                 if tpl['userid'] == userid:
                     tpls.append(tpl)
             tasks = []
             tasklogs = []
             for task in maindb.db.task.list(userid, fields=('id', 'tplid', 'userid', 'note', 'disabled', 'init_env', 'env', 'session', 'retry_count', 'retry_interval', 'last_success', 'success_count',
-                                                        'failed_count', 'last_failed', 'next', 'last_failed_count', 'ctime', 'mtime', 'ontimeflg', 'ontime',  '_groups', 'pushsw', 'newontime'), limit=None):
+                                                            'failed_count', 'last_failed', 'next', 'last_failed_count', 'ctime', 'mtime', 'ontimeflg', 'ontime', '_groups', 'pushsw', 'newontime'), limit=None):
                 if task['userid'] == userid:
                     tasks.append(task)
-                    for tasklog in maindb.db.tasklog.list(taskid = task['id'], fields=('id', "taskid", "success", "ctime", "msg")):
+                    for tasklog in maindb.db.tasklog.list(taskid=task['id'], fields=('id', "taskid", "success", "ctime", "msg")):
                         tasklogs.append(tasklog)
 
             c.close()
             conn.close()
 
-
         except Exception as e:
-            raise Exception("backup database error")
+            raise Exception("backup database error") from e
         print("OK")

chrole.py

@@ -10,16 +10,20 @@
 """
 
 import asyncio
+import logging
 import sys
 
 import db
 
+logger = logging.getLogger(__name__)
+
 
 def usage():
-    print('Usage: python3 %s <email> [role]' % sys.argv[0])
-    print('Example: python3 %s [email protected] admin' % sys.argv[0])
+    print(f'Usage: python3 {sys.argv[0]} <email> [role]')
+    print(f'Example: python3 {sys.argv[0]} [email protected] admin')
     sys.exit(1)
 
+
 async def main():
     email = sys.argv[1]
     role = sys.argv[2] if len(sys.argv) == 3 else ''
@@ -31,9 +35,9 @@ async def main():
         sys.exit(1)
     rowcount = await userdb.mod(user['id'], role=role)
     if rowcount >= 1:
-        print("role of {} changed to {}".format(email, role or '[empty]'))
+        logger.info("role of %s changed to %s", email, role or '[empty]')
     else:
-        print("role of {} not changed".format(email))
+        logger.warning("role of %s not changed", email)
 
 
 if __name__ == '__main__':

db/basedb.py

@@ -22,59 +22,60 @@
 from libs.log import Log
 
 if config.db_type == 'mysql':
-    host=config.mysql.host
-    port=config.mysql.port
-    database=config.mysql.database
-    user=config.mysql.user
-    passwd=config.mysql.passwd
-    auth_plugin=config.mysql.auth_plugin
+    host = config.mysql.host
+    port = config.mysql.port
+    database = config.mysql.database
+    user = config.mysql.user
+    passwd = config.mysql.passwd
+    auth_plugin = config.mysql.auth_plugin
     engine_url = f"mysql+aiomysql://{user}:{passwd}@{host}:{port}/{database}?auth_plugin={auth_plugin}"
     engine = create_async_engine(engine_url,
-                                logging_name = config.sqlalchemy.logging_name,
-                                pool_size = config.sqlalchemy.pool_size,
-                                max_overflow = config.sqlalchemy.max_overflow,
-                                pool_logging_name = config.sqlalchemy.pool_logging_name,
-                                pool_pre_ping = config.sqlalchemy.pool_pre_ping,
-                                pool_recycle = config.sqlalchemy.pool_recycle,
-                                pool_timeout = config.sqlalchemy.pool_timeout,
-                                pool_use_lifo = config.sqlalchemy.pool_use_lifo)
+                                 logging_name=config.sqlalchemy.logging_name,
+                                 pool_size=config.sqlalchemy.pool_size,
+                                 max_overflow=config.sqlalchemy.max_overflow,
+                                 pool_logging_name=config.sqlalchemy.pool_logging_name,
+                                 pool_pre_ping=config.sqlalchemy.pool_pre_ping,
+                                 pool_recycle=config.sqlalchemy.pool_recycle,
+                                 pool_timeout=config.sqlalchemy.pool_timeout,
+                                 pool_use_lifo=config.sqlalchemy.pool_use_lifo)
 elif config.db_type == 'sqlite3':
     engine_url = f"sqlite+aiosqlite:///{config.sqlite3.path}"
     engine = create_async_engine(engine_url,
-                                logging_name = config.sqlalchemy.logging_name,
-                                pool_logging_name = config.sqlalchemy.pool_logging_name,
-                                pool_pre_ping = config.sqlalchemy.pool_pre_ping,
-                                pool_recycle = config.sqlalchemy.pool_recycle )
+                                 logging_name=config.sqlalchemy.logging_name,
+                                 pool_logging_name=config.sqlalchemy.pool_logging_name,
+                                 pool_pre_ping=config.sqlalchemy.pool_pre_ping,
+                                 pool_recycle=config.sqlalchemy.pool_recycle)
     Log('aiosqlite',
         logger_level=config.sqlalchemy.pool_logging_level,
         channel_level=config.sqlalchemy.pool_logging_level).getlogger()
 else:
     raise Exception('db_type must be mysql or sqlite3')
-logger_DB = Log('sqlalchemy',
-                logger_level=config.sqlalchemy.logging_level,
-                channel_level=config.sqlalchemy.logging_level).getlogger()
-logger_DB_Engine = Log(engine.engine.logger,
+logger_db = Log('sqlalchemy',
                 logger_level=config.sqlalchemy.logging_level,
                 channel_level=config.sqlalchemy.logging_level).getlogger()
+logger_db_engine = Log(engine.engine.logger,
+                       logger_level=config.sqlalchemy.logging_level,
+                       channel_level=config.sqlalchemy.logging_level).getlogger()
 if hasattr(engine.pool.logger, 'logger'):
-    logger_DB_POOL = Log(engine.pool.logger.logger,
-                        logger_level=config.sqlalchemy.pool_logging_level,
-                        channel_level=config.sqlalchemy.pool_logging_level).getlogger()
+    logger_db_pool = Log(engine.pool.logger.logger,
+                         logger_level=config.sqlalchemy.pool_logging_level,
+                         channel_level=config.sqlalchemy.pool_logging_level).getlogger()
 else:
-    logger_DB_POOL = Log(engine.pool.logger,
-                        logger_level=config.sqlalchemy.pool_logging_level,
-                        channel_level=config.sqlalchemy.pool_logging_level).getlogger()
+    logger_db_pool = Log(engine.pool.logger,
+                         logger_level=config.sqlalchemy.pool_logging_level,
+                         channel_level=config.sqlalchemy.pool_logging_level).getlogger()
 async_session = async_scoped_session(sessionmaker(engine, class_=AsyncSession, expire_on_commit=False),
                                      scopefunc=current_task)
 BaseDB = declarative_base(bind=engine, name="BaseDB")
 
+
 class AlchemyMixin:
     @property
     def sql_session(self) -> AsyncSession:
         return async_session()
 
     @contextlib.asynccontextmanager
-    async def transaction(self, sql_session:AsyncSession=None):
+    async def transaction(self, sql_session: AsyncSession = None):
         if sql_session is None:
             async with self.sql_session as sql_session:
                 # deepcode ignore AttributeLoadOnNone: sql_session is not None
@@ -86,14 +87,14 @@ async def transaction(self, sql_session:AsyncSession=None):
         else:
             yield sql_session
 
-    async def _execute(self, text:Tuple[str,text], sql_session:AsyncSession=None):
+    async def _execute(self, text: Tuple[str, text], sql_session: AsyncSession = None):
         async with self.transaction(sql_session) as sql_session:
             if isinstance(text, str):
                 text = text.replace(':', r'\:')
             result = await sql_session.execute(text)
             return result
 
-    async def _get(self, stmt: Select, one_or_none=False, first=False, all=True, sql_session:AsyncSession=None):
+    async def _get(self, stmt: Select, one_or_none=False, first=False, all=True, sql_session: AsyncSession = None):
         async with self.transaction(sql_session) as sql_session:
             result: Result = await sql_session.execute(stmt)
             if one_or_none:
@@ -105,7 +106,7 @@ async def _get(self, stmt: Select, one_or_none=False, first=False, all=True, sql
             else:
                 return result
 
-    async def _insert(self, instance, many=False, sql_session:AsyncSession=None):
+    async def _insert(self, instance, many=False, sql_session: AsyncSession = None):
         async with self.transaction(sql_session) as sql_session:
             if many:
                 sql_session.add_all(instance)
@@ -114,24 +115,24 @@ async def _insert(self, instance, many=False, sql_session:AsyncSession=None):
                 await sql_session.flush()
                 return instance.id
 
-    async def _update(self, stmt: Update, sql_session:AsyncSession=None):
+    async def _update(self, stmt: Update, sql_session: AsyncSession = None):
         async with self.transaction(sql_session) as sql_session:
             result: Result = await sql_session.execute(stmt)
             return result.rowcount
 
-    async def _insert_or_update(self, insert_stmt: Insert, sql_session:AsyncSession=None, **kwargs) -> int:
+    async def _insert_or_update(self, insert_stmt: Insert, sql_session: AsyncSession = None, **kwargs) -> int:
         async with self.transaction(sql_session) as sql_session:
             on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(**kwargs)
             result: CursorResult = await sql_session.execute(on_duplicate_key_stmt)
             return result.lastrowid
 
-    async def _delete(self, stmt: Delete, sql_session:AsyncSession=None):
+    async def _delete(self, stmt: Delete, sql_session: AsyncSession = None):
         async with self.transaction(sql_session) as sql_session:
             result: Result = await sql_session.execute(stmt)
             return result.rowcount
 
     @staticmethod
-    def to_dict(result,fields=None):
+    def to_dict(result, fields=None):
         if result is None:
             return result
         if fields is None: