diff --git a/swh/storage/db.py b/swh/storage/db.py index fcac5662..874ccdb7 100644 --- a/swh/storage/db.py +++ b/swh/storage/db.py @@ -1,33 +1,83 @@ # 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 +from contextlib import contextmanager + +TMP_CONTENT_TABLE = 'tmp_content' + class Db: - """DB proxy + """Proxy to the SWH DB, with wrappers around stored procedures """ - def __init__(self, conn): - """create a DB proxy, connecting to the DB + @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: - conn: either a libpq connection string, or an established psycopg2 - connection to the SWH DB + connstring: libpq2 connection string + """ - if isinstance(conn, psycopg2.extensions.connection): - self.conn = conn + 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: - self.conn = psycopg2.connect(conn) + 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 cursor(self): - return self.conn.cursor() + # @stored_procedure('swh_content_mktemp') + def content_mktemp(self, cur=None): + self._cursor(cur).execute('SELECT swh_content_mktemp()') - def commit(self): - return self.conn.commit() + 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 rollback(self): - return self.conn.rollback() + # @stored_procedure('swh_content_add') + def content_add_from_temp(self, cur): + self._cursor(cur).execute('SELECT swh_content_add()') diff --git a/swh/storage/storage.py b/swh/storage/storage.py index 8d0ccbc3..ffeb6484 100644 --- a/swh/storage/storage.py +++ b/swh/storage/storage.py @@ -1,71 +1,72 @@ # 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 -TMP_CONTENT_TABLE = 'tmp_content' - 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 an already - established psycopg2 connection to the SWH DB + db_conn: either a libpq connection string, or a psycopg2 connection obj_root: path to the root of the object storage """ - self.db = Db(db_conn) + if isinstance(db_conn, psycopg2.extensions.connection): + self.db = Db(db_conn) + else: + self.db = Db.connect(db_conn) + self.objstorage = ObjStorage(obj_root) 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 - one key for each checksum algorithm in swh.core.hashutil(ALGORITHMS), mapped to the corresponding checksum """ - with self.db.cursor() as c: + db = self.db + with db.transaction() as cur: # create temporary table for metadata injection - c.execute('SELECT swh_content_mktemp()') + db.content_mktemp(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) - c.copy_from(f, TMP_CONTENT_TABLE, - columns=('sha1', 'sha1_git', 'sha256', 'length')) - - # move metadata in place and save - c.execute('SELECT swh_content_add()') + db.content_copy_to_temp(f, cur) - self.db.commit() + # 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 1333abb3..95aad066 100644 --- a/swh/storage/tests/test_db.py +++ b/swh/storage/tests/test_db.py @@ -1,30 +1,40 @@ # 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): - @istest - def connect(self): - self.assertFalse(self.conn.closed) + def setUp(self): + super().setUp() + self.db = Db(self.conn) + + def tearDown(self): + self.db.conn.close() + super().tearDown() @istest - def insert_content(self): - self.cursor.execute( - 'INSERT INTO content (sha1, sha1_git, sha256, length, status) ' + - 'VALUES (%s, %s, %s, %s, %s)', - ('34973274ccef6ab4dfaaf86599792fa9c3fe4689', - 'd81cc0710eb6cf9efd5b920a8453e1e07157b6cd', - '673650f936cb3b0a2f93ce09d81be10748b1b203' - 'c19e8176b4eefc1964a0cf3a', - 3, 'visible')) + def add_content(self): + cur = self.cursor + sha1 = '34973274ccef6ab4dfaaf86599792fa9c3fe4689' + self.db.content_mktemp(cur) + self.db.content_copy_to_temp(StringIO( + sha1 + '\t' + 'd81cc0710eb6cf9efd5b920a8453e1e07157b6cd\t' + '673650f936cb3b0a2f93ce09d81be10748b1b203' + 'c19e8176b4eefc1964a0cf3a\t' '3\n'), 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,))