Software Engineering/Python (15) 썸네일형 리스트형 python logging 설정 예제 my_logger.pyimport osimport loggingfrom logging.handlers import RotatingFileHandlerclass ImmediateFlushHandler(RotatingFileHandler): def emit(self, log): super().emit(log) self.flush()class MyLoggingHelper(): def __init__(self, root_path: str): logger = logging.getLogger() logger.setLevel(logging.INFO) # console handler #self.ch = logging.StreamHan.. sample project 연습 (1) 프로젝트 트리 (2) main.pyimport timefrom app_control import AppControlParameterfrom my_timer import MyTimerfrom db_manager import DatabaseManagerfrom event_manager import EventItem, EventMatcher, EventPusherdef process(): control_para = AppControlParameter("my-servicename") control_timer = MyTimer() control_timer.start() reload_timer = MyTimer() reload_timer.start() db = Database.. timer 유틸리티 클래스 import timeclass MyTimer: def __init__(self): self.start_time = None def start(self): self.start_time = time.time() def is_over_sec(self, expected_time_sec: int) -> bool: if self.start_time is None: return False elapsed_time = time.time() - self.start_time return elapsed_time >= expected_time_sec def is_over_min(self, .. pydantic을 사용하여 환경변수 로딩하기 아래와 같이 세팅 클래스를 만들고 인스턴수화 한 후, 필요한 파일에서 호출하여 사용한다.Config 내부 클래스의 'env_file' 속성을 설정하면 해당 클래스를 인스턴스화 할때 자동으로 환경 변수를 로딩한다. from pydantic import BaseSettingsclass Settings(BaseSettings): DATABASE_URL: str KAFKA_BOOTSTRAP_SERVERS: str KAFKA_TOPIC: str EXTERNAL_API_URL: str class Config: env_file = ".env"settings = Settings() 아래 처럼 호출한 후 전역 변수 처럼 사용 가능하다. from app.config.settings .. fastapi post 요청으로 logging 레벨 변경하기 from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelimport loggingapp = FastAPI()# 기본 로깅 설정logging.basicConfig(level=logging.INFO)logger = logging.getLogger("app_logger")# 요청 바디 모델 정의class DebugLevel(BaseModel): level: str# 디버깅 레벨 맵핑LOG_LEVELS = { "CRITICAL": logging.CRITICAL, "ERROR": logging.ERROR, "WARNING": logging.WARNING, "INFO": logging.INFO, "DEBUG":.. fastapi get 요청으로 app 종료하기 * 어플리케이션 종료에 os._exit 또는 sys.exit(0)를 사용할 수 있음 * sys.exit()는 현재 실행 중인 스레드를 종료하는 함수로 내부적으로 SystemExit 예외를 발생시켜 종료를 처리함해당 예외를 핸들링하는 핸들러가 있다면 해당 핸들러가 실행되고, 일반적으로는 finally 블록이나 atexit 모듈을 사용하여 등록된 종료 핸들러가 실행된다. 즉, 연관된 이벤트 핸들러가 동작하므로 정상적인 종료 절차를 밟고 종료할 때 사용한다. * os._exit(0) 현재 실행 중인 프로세스를 즉시 종료한느 함수로 운영체제 레벨에서 프로세스를 종료한다.실행 즉시 프로세스가 종료되며, finally 블록이나 atexit 모듈을 사용하여 등록된 핸들러가 실행되지 않음. 즉시 종료가 필요한 경우나.. pymysql package 사용 예제 (1) 기본 용어/개념* connection: python과 mariadb를 연결하는 객체import pymysqlconnection = pymysql.connect(host='host', user='user', password='password', databse='dbname') * cursor: 쿼리를 실행하고 결과를 가져오는데 사용되는 객체, SQL 문을 실행하고 결과를 처리할 수 있게 지원한다.cursor = connection.cursor() * query: SQL 구문을 작성하고 cursor를 사용하여 실행한다.cursor.execute("SELECT * FROM table_name") * commit: 데이터베이스 트랜젝션을 확정하는 작업으로 데이터 변경이 필요한 insert, update,.. 이전 1 2 다음
