python pyQt5활용하여 fastapi 로그 찍기

소개

 

안녕하세요 오늘은 fastAPI 로그를 pyQt5 ui를 사용하여 표시하는 작업을 진행할 예정입니다.

 

기존에는 pyinstaller를 사용하여 command.exe 와 log파일에 로그기록을 남겨서 사용하고있었습니다.

 

cmd 에 로그를 남기는 이유는 실시간으로 로그를 확인하고 , 프로그램 종료또한 cmd 가 종료되면 종료할수 있게 하기 위해서 였지만 여러가지 문제로 cmd 를 숨긴 상태로 배포하기로 하여 진행 하게 되었습니다.

 

구현 하면서 가장 힘들었던점은 pyQt5 gui 에 log를 표시할때 충돌이 발생하여 프로그램이 죽는 점이였습니다.

 

위 문제를 pyQt 커스텀 시그널을 사용하여 해결하였습니다.

 

간단하게 리뷰 진행할게요 

 

코드

app.py : 최초 실행을 담당하는 main 역활

webServer.py :  fastAPI관련 로직

looger.py : 로그 관련 클래스  (  싱글톤으로 되어있음 )

 

 

app.py

import traceback
from PyQt5.QtWidgets import QApplication, QWidget , QPushButton, QVBoxLayout  , QTextEdit 
from PyQt5.QtCore import *
import sys
from webServer import  app 
from logger import customLogger
import uvicorn


class newFastAPI(QThread):
    def run(self):
        uvicorn.run(app,host='localhost',port=8080)
        

class CustomSignal(QObject):
    signal = pyqtSignal(str) #반드시 클래스 변수로 선언할 것

    def run(self,text):
        # text = "emit으로 전달"
        self.signal.emit(text) #customFunc 메서드 실행시 signal의 emit 메서드사용

class MyApp(QWidget):

    def __init__(self):
        super().__init__()
        self.initUI()

    def initUI(self):
        self.setWindowTitle(' pyQt5 & fastAPI log Test')
        self.move(300, 300)
        self.resize(400, 200)
        # self.setWindowFlags(Qt.CoverWindow)
        
        self.tb = QTextEdit()
        self.tb.setReadOnly(True)
        self.tb.append('시작')

        self.customsignal = CustomSignal() #Mysignal 클래스의 객체 선언
        self.customsignal.signal.connect(self.append_text) #객체에 대한시그널 및 슬롯 설정


        self.clear_btn = QPushButton('Clear')
        self.clear_btn.pressed.connect(self.clear_text)
        
        vbox = QVBoxLayout()
        vbox.addWidget(self.tb, 0)
        vbox.addWidget(self.clear_btn, 1)
        self.setLayout(vbox)
        self.setGeometry(300, 300, 1000, 300)
        
    @pyqtSlot(str)    
    def append_text(self,text):
        self.tb.append(text)

    def clear_text(self):
        self.tb.clear()
    


if __name__ == '__main__':
    try:
        Q = QApplication(sys.argv)
        brower = MyApp()
        brower.show()
        log = customLogger.getInstance()
        log.setMyApp(brower)

        api = newFastAPI()
        api.start()
        sys.exit(Q.exec_())

    except:
        log.showLog('=============================  에러 발생','error')
        log.showLog(traceback.format_exc(),'error')

 

webServer,py

from fastapi import FastAPI  ,Query
from fastapi.middleware.cors import CORSMiddleware
from logger import customLogger

log = customLogger.getInstance()

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

#test
@app.get('/api/test1')
def test1(num: int = Query(default=1)):
    for i  in range(num):
        log.showLog(f'{i} 번째 로그 입니다.')
    return {
        'code':200
    }

 

logger,py

 

from datetime import datetime
from logging import handlers
import logging
import os



class customLogger:
    _instance = None
    def __init__(self):
        if not customLogger._instance:
            print('__init__ method called but nothing is created')
            self.setting()
        else:
            print('instance already created:', self.getInstance())
 
    @classmethod
    def getInstance(cls):
        if not cls._instance:
            cls._instance = customLogger()
        return cls._instance

    def setting(self):
        self.myApp = None
        self.createFolder('./logs/')
        # looger 생성
        self.logger = logging.getLogger()
        #  로그 파일 날짜
        date = datetime.today().strftime("%Y%m%d")
        # 저장되는 파일 주기
        file_handler = handlers.TimedRotatingFileHandler(filename=f'./logs/{date}.log',  when='D',  encoding='utf-8'  )
        # 저장되는 파일 형식
        formatter = logging.Formatter(u'%(asctime)s [%(levelname)8s] %(message)s') 
        # 파일 형식 설정
        file_handler.setFormatter(formatter)
        # logger와 연결 
        self.logger.addHandler(file_handler)
    
    def createFolder(self,directory):
        try:
            if not os.path.exists(directory):
                os.makedirs(directory)
        except OSError:
            print ('Error: Creating directory. ' +  directory)
            
    def showLog(self,message):
        self.logger.info(f'{message}')
        if self.myApp != None:
            self.myApp.customsignal.run(message)
        
    def setMyApp(self,myApp):
        self.myApp = myApp

 

상세 로직 flow

 

1. app.py 실행

2. pyQt5 실행

3. log 생성

4. log에 pyQt 전달

5. fastAPI 스레드로 생성

 

문제점

위에서도 말씀드렸지만 logger.py 에서  myApp 의 append_text를 직접 접근하여 tb에 append 할 경우 간혈적으로 프로그램이 죽습니다. 

 

(실제 저희 프로그램은 threading.Thread 가 사용되고있는데 찾아보니까 QThread를 사용하면 된다고 해서 해봣는데, 테스트 단까지는 안죽고 됬는데 실제로 서비스 배포하니까 계속 죽더라고요 .)

 

저는 아래 메시지가 많이 발생햇었습니다.

Cannot queue arguments of type 'QTextCursor'(Make sure 'QTextCursor' is registered using qRegisterMetaType().

 

다만 프로그램이 죽을때 위 메시지를 표시하고 죽지는 않습니다.

 

제가 예전에 안드로이드를 조금 다뤄봤는데 그때 아마 java 단에서 스레드를 생성해서 ui에 접근하면 프로그램이 죽었던 부분이 생각나서 그부분 위주로 찾아 해결 했습니다.

 

 

해결방안 

 

pyQt 는 시그널과 슬롯이라는 개념이 존재합니다.

 

자세한 정보는 아래 참조하시면 될거같아요. 

 

https://ybworld.tistory.com/109

[Python/파이썬] PyQt5 - 시그널(Signal)과 슬롯(Slot) 개념

 

위 내용처럼 커스텀 시그널을 사용하였습니다.

logger 에서 로그 발생시  self.customsignal.run 에 해당 로그를 전송하고  CustomSignal의 self.signal.emit 에 해당 내용을 넣어주니까 실질적으로 ui ( tb) 에 append 가가능하고 에러가 발생하지않앗습니다.

 

 

 

결과

 

python app.py 실행

 

 

postman 으로 http 요청 

 

 

 

최종 결과

 

 

 

이상입니다.

 

 

후기

 

pyQt5는 QThread 를 상속받아 동작합니다. 

QThread가 threading.Thread를 죽인다는 글도 보긴햇는데 공문까지는 가보지않앗습니다.

커스텀 시그널 사용안하고 그냥 되는 분도 있는거같은데 , 저희는 실제로 스레드를 여러개 돌려서 그런지 무조건죽더라고요.