烦恼一般都是想太多了。

0%

Python编码中的一些实用片段

用以记录,经常用到,懒得再看。

logging 日志

def init_logger(name):
handler = TimedRotatingFileHandler(LOGFILE % name, 'D', 1, 30)
shandlder = logging.StreamHandler()

fmt = "%(asctime)-15s %(levelname)s %(filename)s %(lineno)d %(process)d %(message)s"
formatter = logging.Formatter(fmt)

handler.setFormatter(formatter)
shandlder.setFormatter(formatter)

logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG if DEBUG else logging.INFO)
logger.addHandler(handler)
logger.addHandler(shandlder)

return logger

daemon 后台运行

def daemon(func, pidfile):
if os.path.exists(pidfile):
print('程序已经启动')
os._exit(0)
pid = os.fork()
if pid < 0:
print('创建子进程错误')
os._exit(1)
if pid > 0:
os._exit(0)
# os.chdir('/')
# os.umask(0o22)

ppid = os.fork()
if ppid < 0:
print('创建子进程错误')
os._exit(1)
if ppid == 0:
with open(pidfile, 'w+') as f:
f.write(str(os.getpid()))
f0 = open('/dev/null', 'rb', 0)
f1 = open('/dev/null', 'ab', 0)
f2 = open('/dev/null', 'ab', 0)
os.dup2(f0.fileno(), sys.stdin.fileno())
os.dup2(f1.fileno(), sys.stdout.fileno())
os.dup2(f2.fileno(), sys.stderr.fileno())
atexit.register(os.remove, pidfile)
signal.signal(signal.SIGTERM, handler)
signal.signal(signal.SIGINT, handler)
func()
print('程序已启动', ppid)

HTTP 服务器

class Handler(BaseHTTPRequestHandler):
def do_POST(self):
length = int(self.headers['content-length'])
if length < 5:
self.failed('无任何数据')
return

data = self.rfile.read(length)
params = json.loads(data.decode())
if len(params) == 0:
self.failed('未传递参数')
return
if len(params) < 2:
self.failed('参数不足 2 个')
return
if 'ywlxdm' not in params:
self.failed('ywlxdm 参数不存在')
return
if 'comName' not in params:
self.failed('企业名称不存在')
return
comName, ywlxdm = params.values()
if self.path == '/do_print':
PRTMGR.check_status()
if PRTMGR.is_error:
self.failed(PRTMGR.errorstateinfo)
return
self.success('打印请求发送成功')
task = Task(comName, ywlxdm)
TASKS[Task.key(comName, ywlxdm)] = task
TASKQUEUE.put(task)
return
if self.path == '/query':
key = Task.key(comName, ywlxdm)
if key in TASKS:
task = TASKS[key]
self.success('成功', task.json())
if task.status > 10:
del TASKS[key]
return
self.failed('无此进行中的任务信息')
return
APISERVER = ThreadingHTTPServer(HOST, Handler)
if __name__ == '__main__':
APISERVER.serve_forever()

注意这是一个多线程的服务器。如果不想使用多线程,就把 ThreadingHTTPServer 换成 HTTPServer

class ThreadingHTTPServer(socketserver.ThreadingMixIn, HTTPServer):
daemon_threads = True

这个多线程的实现,实际就是在 socketserver.ThreadingMixIn 中重写了 process_request 开启一个新线程来处理。