Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/storage.py
| # Copyright (C) 2015-2017 The Software Heritage developers | # Copyright (C) 2015-2017 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from collections import defaultdict | from collections import defaultdict | ||||
| import datetime | import datetime | ||||
| import itertools | import itertools | ||||
| import json | import json | ||||
| import dateutil.parser | import dateutil.parser | ||||
| import psycopg2 | import psycopg2 | ||||
| import psycopg2.pool | |||||
| from . import converters | from . import converters | ||||
| from .common import db_transaction_generator, db_transaction | from .common import db_transaction_generator, db_transaction | ||||
| from .db import Db | from .db import Db | ||||
| from .exc import StorageDBError | from .exc import StorageDBError | ||||
| from .algos import diff | from .algos import diff | ||||
| from swh.model.hashutil import ALGORITHMS | from swh.model.hashutil import ALGORITHMS | ||||
| from swh.objstorage import get_objstorage | from swh.objstorage import get_objstorage | ||||
| from swh.objstorage.exc import ObjNotFoundError | from swh.objstorage.exc import ObjNotFoundError | ||||
| # Max block size of contents to return | # Max block size of contents to return | ||||
| BULK_BLOCK_CONTENT_LEN_MAX = 10000 | BULK_BLOCK_CONTENT_LEN_MAX = 10000 | ||||
| CONTENT_HASH_KEYS = ['sha1', 'sha1_git', 'sha256', 'blake2s256'] | CONTENT_HASH_KEYS = ['sha1', 'sha1_git', 'sha256', 'blake2s256'] | ||||
| class Storage(): | class Storage(): | ||||
| """SWH storage proxy, encompassing DB and object storage | """SWH storage proxy, encompassing DB and object storage | ||||
| """ | """ | ||||
| def __init__(self, db, objstorage): | def __init__(self, db, objstorage, min_pool_conns=1, max_pool_conns=10): | ||||
| """ | """ | ||||
| Args: | Args: | ||||
| db_conn: either a libpq connection string, or a psycopg2 connection | db_conn: either a libpq connection string, or a psycopg2 connection | ||||
| obj_root: path to the root of the object storage | obj_root: path to the root of the object storage | ||||
| """ | """ | ||||
| try: | try: | ||||
| if isinstance(db, psycopg2.extensions.connection): | if isinstance(db, psycopg2.extensions.connection): | ||||
| self._pool = None | |||||
| self._db = Db(db) | self._db = Db(db) | ||||
| else: | else: | ||||
| self._db = Db.connect(db) | self._pool = psycopg2.pool.ThreadedConnectionPool( | ||||
| min_pool_conns, max_pool_conns, db | |||||
| ) | |||||
| self._db = None | |||||
| except psycopg2.OperationalError as e: | except psycopg2.OperationalError as e: | ||||
| raise StorageDBError(e) | raise StorageDBError(e) | ||||
| self.objstorage = get_objstorage(**objstorage) | self.objstorage = get_objstorage(**objstorage) | ||||
| def get_db(self): | def get_db(self): | ||||
| if self._db: | |||||
| return self._db | return self._db | ||||
| else: | |||||
| return Db.from_pool(self._pool) | |||||
| def check_config(self, *, check_write): | def check_config(self, *, check_write): | ||||
| """Check that the storage is configured and ready to go.""" | """Check that the storage is configured and ready to go.""" | ||||
| if not self.objstorage.check_config(check_write=check_write): | if not self.objstorage.check_config(check_write=check_write): | ||||
| return False | return False | ||||
| # Check permissions on one of the tables | # Check permissions on one of the tables | ||||
| ▲ Show 20 Lines • Show All 1,505 Lines • Show Last 20 Lines | |||||