75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
# This file is part of the Open Parts Database software
|
|
# Copyright (C) 2022 Valentin Lorentz
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it under the
|
|
# terms of the GNU Affero General Public License version 3, as published by the
|
|
# Free Software Foundation.
|
|
#
|
|
# 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 Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License along with
|
|
# this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
"""
|
|
Abstraction over the postgresql database used by OPDB
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import typing
|
|
|
|
import psycopg
|
|
|
|
from . import models
|
|
|
|
|
|
class Db:
|
|
"""
|
|
Abstraction over the postgresql database used by OPDB
|
|
"""
|
|
|
|
def __init__(self, conn: psycopg.Connection):
|
|
self.conn = conn
|
|
|
|
@classmethod
|
|
@contextlib.contextmanager
|
|
def open(cls, dsn: str) -> typing.Iterator[Db]:
|
|
"""
|
|
Context manager, which yields a :class:`Db` object given a libpq connection
|
|
string (DSN)
|
|
"""
|
|
with psycopg.connect(dsn) as conn:
|
|
yield Db(conn)
|
|
|
|
def init(self) -> None:
|
|
"""
|
|
Initializes the schema for the connected database.
|
|
"""
|
|
with self.conn.cursor() as cur:
|
|
for name in dir(models):
|
|
cls = getattr(models, name)
|
|
if hasattr(cls, "TABLE"):
|
|
cur.execute(cls.db_schema())
|
|
|
|
def get_last_web_page_snapshot(
|
|
self, url: str
|
|
) -> typing.Optional[models.WebPageSnapshot]:
|
|
"""
|
|
Returns the last snapshot of the given IRI.
|
|
"""
|
|
snapshots = models.WebPageSnapshot.select(
|
|
self.conn, "WHERE url=%s ORDER BY snapshot_date DESC LIMIT 1", (url,)
|
|
)
|
|
return next(snapshots, None)
|
|
|
|
def add_web_page_snapshots(
|
|
self, snapshots: typing.Iterable[models.WebPageSnapshot]
|
|
) -> None:
|
|
"""
|
|
Stores new snapshots of web pages to the database.
|
|
"""
|
|
models.WebPageSnapshot.copy_to_db(self.conn, snapshots)
|