mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-06-27 22:31:43 +00:00
Evgenii Alekseev
137d62e2f8
* implement log storage at backend * handle process id during removal. During one process we can write logs from different packages in different times (e.g. check and update later) and we would like to store all logs belong to the same process * set package context in main functions * implement logs support in interface * filter out logs posting http logs * add timestamp to log records * hide getting logs under reporter permission List of breaking changes: * `ahriman.core.lazy_logging.LazyLogging` has been renamed to `ahriman.core.log.LazyLogging` * `ahriman.core.configuration.Configuration.from_path` does not have `quiet` attribute now * `ahriman.core.configuration.Configuration` class does not have `load_logging` method now * `ahriman.core.status.client.Client.load` requires `report` argument now
82 lines
2.4 KiB
Python
82 lines
2.4 KiB
Python
#
|
|
# Copyright (c) 2021-2022 ahriman team.
|
|
#
|
|
# This file is part of ahriman
|
|
# (see https://github.com/arcan1s/ahriman).
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
import sqlite3
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Tuple, TypeVar, Callable
|
|
|
|
from ahriman.core.log import LazyLogging
|
|
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class Operations(LazyLogging):
|
|
"""
|
|
base operation class
|
|
|
|
Attributes:
|
|
path(Path): path to the database file
|
|
"""
|
|
|
|
def __init__(self, path: Path) -> None:
|
|
"""
|
|
default constructor
|
|
|
|
Args:
|
|
path(Path): path to the database file
|
|
"""
|
|
self.path = path
|
|
|
|
@staticmethod
|
|
def factory(cursor: sqlite3.Cursor, row: Tuple[Any, ...]) -> Dict[str, Any]:
|
|
"""
|
|
dictionary factory based on official documentation
|
|
|
|
Args:
|
|
cursor(Cursor): cursor descriptor
|
|
row(Tuple[Any, ...]): fetched row
|
|
|
|
Returns:
|
|
Dict[str, Any]: row converted to dictionary
|
|
"""
|
|
result = {}
|
|
for index, column in enumerate(cursor.description):
|
|
result[column[0]] = row[index]
|
|
return result
|
|
|
|
def with_connection(self, query: Callable[[sqlite3.Connection], T], *, commit: bool = False) -> T:
|
|
"""
|
|
perform operation in connection
|
|
|
|
Args:
|
|
query(Callable[[Connection], T]): function to be called with connection
|
|
commit(bool, optional): if True commit() will be called on success (Default value = False)
|
|
|
|
Returns:
|
|
T: result of the ``query`` call
|
|
"""
|
|
with sqlite3.connect(self.path, detect_types=sqlite3.PARSE_DECLTYPES) as connection:
|
|
connection.row_factory = self.factory
|
|
result = query(connection)
|
|
if commit:
|
|
connection.commit()
|
|
return result
|