تو این پست میخوایم یک لوگر درست کنیم که لوگ ها رو تو دیتابیس ذخیره کنه.
برای اینکار ما از SQLAlchemy استفاده میکنیم پیشنهاد میکنم این پستها رو بخونید
تو این آموزش از SQLite استفاده میکنیم
اول اتصال رو آماده میکنیم
1 2 3 4 5 6 7 8 |
from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker Base = declarative_base() db_engine = create_engine('sqlite:///db.sqlite') db_session_handler = sessionmaker(db_engine) |
و کلاس مدل لوگر
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
from datetime import datetime from sqlalchemy import Column from sqlalchemy import Integer from sqlalchemy import String from sqlalchemy import Text from sqlalchemy import Datetime from sqlalchemy.orm import validates class Log(Base): __tablename__ = "logs" id = Column(Integer, primary_key=True, nullable=False) levelname = Column(String, nullable=False) msg = Column(Text, nullable=False) time = Column(DateTime, nullable=False, default=datetime.now) @validates("levelname") def validate_levelname(self, key, value): if value not in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"): raise APPModelError("bad input for log levelname") return value |
حالا جدول لوگ ها رو میسازیم
1 2 3 4 |
# this won't raise error if table already exists, this is achieved by set `checkfirst=True` Log.__table__.create(bind=db_engine, checkfirst=True) # or Base.metdata.create_all(bind=db_engine) |
و لوگر اصلی
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
from logging import Handler class DatabaseHandler(Handler): def emit(self, record): dbs = db_session_handler() try: row = Log() row.levelname = record.levelname row.msg = record.msg % record.args if record.args else str(record.msg) dbs.add(row) dbs.commit() except Exception as err: dbs.rollback() raise finally: dbs.close() |
و تست لوگر
1 2 3 4 5 6 |
logger = logging.getLogger(__name__) logger.setLevel('DEBUG') database_handle = DatabaseHandler() database_handler.setLevel('DEBUG') logger.addHandler(database_handler) logger.debug('test') |