diff --git a/swh/storage/db.py b/swh/storage/db.py index de77aeb3..f36f534c 100644 --- a/swh/storage/db.py +++ b/swh/storage/db.py @@ -1,97 +1,103 @@ # Copyright (C) 2015 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import psycopg2 +import tempfile from contextlib import contextmanager TMP_CONTENT_TABLE = 'tmp_content' def stored_procedure(stored_proc): """decorator to execute remote stored procedure, specified as argument Generally, the body of the decorated function should be empty. If it is not, the stored procedure will be executed first; the function body then. """ def wrap(meth): def _meth(self, *args, **kwargs): cur = kwargs.get('cur', None) self._cursor(cur).execute('SELECT %s()' % stored_proc) meth(self, *args, **kwargs) return _meth return wrap class Db: """Proxy to the SWH DB, with wrappers around stored procedures """ @classmethod def connect(cls, *args, **kwargs): """factory method to create a DB proxy Accepts all arguments of psycopg2.connect; only some specific possibilities are reported below. Args: connstring: libpq2 connection string """ conn = psycopg2.connect(*args, **kwargs) return cls(conn) def _cursor(self, cur_arg): """get a cursor: from cur_arg if given, or a fresh one otherwise meant to avoid boilerplate if/then/else in methods that proxy stored procedures """ if cur_arg is not None: return cur_arg # elif self.cur is not None: # return self.cur else: return self.conn.cursor() def __init__(self, conn): """create a DB proxy Args: conn: psycopg2 connection to the SWH DB """ self.conn = conn @contextmanager def transaction(self): """context manager to execute within a DB transaction Yields: a psycopg2 cursor """ with self.conn.cursor() as cur: try: yield cur self.conn.commit() except: if not self.conn.closed: self.conn.rollback() raise def mktemp(self, tblname, cur=None): self._cursor(cur).execute('SELECT swh_mktemp(%s)', (tblname,)) - def content_copy_to_temp(self, fileobj, cur=None): - self._cursor(cur) \ - .copy_from(fileobj, TMP_CONTENT_TABLE, - columns=('sha1', 'sha1_git', 'sha256', 'length')) + def copy_to(self, items, tblname, columns, cur=None, item_cb=None): + with tempfile.TemporaryFile('w+') as f: + for d in items: + if item_cb is not None: + item_cb(d) + line = '\t'.join([str(d[k]) for k in columns]) + '\n' + f.write(line) + f.seek(0) + self._cursor(cur).copy_from(f, tblname, columns=columns) @stored_procedure('swh_content_add') def content_add_from_temp(self, cur=None): pass diff --git a/swh/storage/storage.py b/swh/storage/storage.py index 34113941..ed503a58 100644 --- a/swh/storage/storage.py +++ b/swh/storage/storage.py @@ -1,89 +1,81 @@ # Copyright (C) 2015 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import psycopg2 import tempfile from .db import Db from .objstorage import ObjStorage def db_transaction(meth): """decorator to execute Storage methods within DB transactions Decorated methods will have access to the following attributes: self.cur: psycopg2 DB cursor """ def _meth(self, *args, **kwargs): with self.db.transaction() as cur: try: self.cur = cur meth(self, *args, **kwargs) finally: self.cur = None return _meth class Storage(): """SWH storage proxy, encompassing DB and object storage """ def __init__(self, db_conn, obj_root): """ Args: db_conn: either a libpq connection string, or a psycopg2 connection obj_root: path to the root of the object storage """ if isinstance(db_conn, psycopg2.extensions.connection): self.db = Db(db_conn) else: self.db = Db.connect(db_conn) self.objstorage = ObjStorage(obj_root) @db_transaction def add_content(self, content): """Add content blobs to the storage Note: in case of DB errors, objects might have already been added to the object storage and will not be removed. Since addition to the object storage is idempotent, that should not be a problem. Args: content: iterable of dictionaries representing individual pieces of content to add. Each dictionary has the following keys: - data (bytes): the actual content + - length (int): content length - one key for each checksum algorithm in swh.core.hashutil(ALGORITHMS), mapped to the corresponding checksum """ (db, cur) = (self.db, self.cur) # create temporary table for metadata injection db.mktemp('content', cur) - with tempfile.TemporaryFile('w+') as f: - # prepare tempfile for metadata COPY + add content data to - # object storage - for cont in content: - cont['length'] = len(cont['data']) - line = '\t'.join([cont['sha1'], cont['sha1_git'], - cont['sha256'], str(len(cont['data']))])\ - + '\n' - f.write(line) - self.objstorage.add_bytes(cont['data'], - obj_id=cont['sha1']) - - # COPY metadata to temporary table - f.seek(0) - db.content_copy_to_temp(f, cur) + def add_to_objstorage(cont): + self.objstorage.add_bytes(cont['data'], obj_id=cont['sha1']) + + db.copy_to(content, 'tmp_content', + ['sha1', 'sha1_git', 'sha256', 'length'], + cur, item_cb=add_to_objstorage) # move metadata in place db.content_add_from_temp(cur) db.conn.commit() diff --git a/swh/storage/tests/test_db.py b/swh/storage/tests/test_db.py index 1668c2cd..12680931 100644 --- a/swh/storage/tests/test_db.py +++ b/swh/storage/tests/test_db.py @@ -1,40 +1,43 @@ # Copyright (C) 2015 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import unittest -from io import StringIO from nose.tools import istest from nose.plugins.attrib import attr from .db_testing import DbTestFixture from swh.storage.db import Db @attr('db') class TestDb(DbTestFixture, unittest.TestCase): def setUp(self): super().setUp() self.db = Db(self.conn) def tearDown(self): self.db.conn.close() super().tearDown() @istest def add_content(self): cur = self.cursor sha1 = '34973274ccef6ab4dfaaf86599792fa9c3fe4689' self.db.mktemp('content', cur) - self.db.content_copy_to_temp(StringIO( - sha1 + '\t' - 'd81cc0710eb6cf9efd5b920a8453e1e07157b6cd\t' - '673650f936cb3b0a2f93ce09d81be10748b1b203' - 'c19e8176b4eefc1964a0cf3a\t' '3\n'), cur) + self.db.copy_to([{ + 'sha1': sha1, + 'sha1_git': 'd81cc0710eb6cf9efd5b920a8453e1e07157b6cd', + 'sha256': '673650f936cb3b0a2f93ce09d81be10748b1b203' + 'c19e8176b4eefc1964a0cf3a', + 'length': 3}], + 'tmp_content', + ['sha1', 'sha1_git', 'sha256', 'length'], + cur) self.db.content_add_from_temp(cur) self.cursor.execute('SELECT sha1 FROM content WHERE sha1 = %s', (sha1,)) self.assertEqual(self.cursor.fetchone(), (sha1,)) diff --git a/swh/storage/tests/test_storage.py b/swh/storage/tests/test_storage.py index daa15c56..94c7f3ae 100644 --- a/swh/storage/tests/test_storage.py +++ b/swh/storage/tests/test_storage.py @@ -1,41 +1,45 @@ # Copyright (C) 2015 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import shutil import tempfile import unittest from nose.tools import istest from nose.plugins.attrib import attr from .db_testing import DbTestFixture from swh.storage import Storage @attr('db') class TestStorage(DbTestFixture, unittest.TestCase): def setUp(self): super().setUp() self.objroot = tempfile.mkdtemp() self.storage = Storage(self.conn, self.objroot) def tearDown(self): shutil.rmtree(self.objroot) super().tearDown() @istest def add_content(self): cont = { 'data': b'42\n', + 'length': 3, 'sha1': '34973274ccef6ab4dfaaf86599792fa9c3fe4689', 'sha1_git': 'd81cc0710eb6cf9efd5b920a8453e1e07157b6cd', 'sha256': '673650f936cb3b0a2f93ce09d81be10748b1b203c19e8176b4eefc1964a0cf3a' # NOQA } self.storage.add_content([cont]) self.assertIn(cont['sha1'], self.storage.objstorage) - self.cursor.execute('SELECT sha1 FROM content WHERE sha1 = %s', + self.cursor.execute('SELECT sha1, sha1_git, sha256, length, status' + ' FROM content WHERE sha1 = %s', (cont['sha1'],)) - self.assertEqual(self.cursor.fetchone(), (cont['sha1'],)) + self.assertEqual(self.cursor.fetchone(), + (cont['sha1'], cont['sha1_git'], cont['sha256'], + cont['length'], 'visible'))