Switched from raw sqlite3 to SQLAlchemy.

hashdb.py

@@ -7,36 +7,35 @@ import hashlib
 import sqlite3
 import uuid
 
+import sqlalchemy
+import sqlalchemy.ext.declarative
+import sqlalchemy.orm
+
 # TODO: l o g g i n g
 HASH_CHUNK_SIZE = 10485760 # 10mb
-SQL_AND = ' AND '
+_db_session_maker = sqlalchemy.orm.sessionmaker()
-SQL_OR = ' OR '
 
-# TODO: Figure out how to do some kind of type checking for these named tuples.
-# TODO: Switch these over to using traditional classes.
-# TODO: UUID generation/editing is probably best done in here.
 ImageData = collections.namedtuple('ImageData', 'UUID, sha1sum, filename, release_group')
 ReleaseGroupData = collections.namedtuple('ReleaseGroupData', 'UUID, name, platform')
 DatData = collections.namedtuple('DatData', 'UUID, name, website, version, image_list')
 
+
 def _uuidgen():
     return str(uuid.uuid4())
 
-@dataclasses.dataclass
+_SQLBase = sqlalchemy.ext.declarative.declarative_base()
-class BaseMetaData:
-    #TODO: See below
-    '''
-    Move the uuid property in here once this bug gets fixed.
-       https://bugs.python.org/issue36077
 
-    Inheriting properties with default values is currently kind of broken.
+class Platform(_SQLBase):
-    '''
+    __tablename__ = 'platforms'
+    id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.Sequence('platform_id_sequence'),
+                           primary_key=True)
+    uuid = sqlalchemy.Column(sqlalchemy.String, nullable=False, default=_uuidgen)
+    fullname = sqlalchemy.Column(sqlalchemy.String, nullable=False)
+    shortcode = sqlalchemy.Column(sqlalchemy.String, unique=True, nullable=False)
 
-@dataclasses.dataclass
+    def __repr__(self):
-class PlatformData(BaseMetaData):
+        return 'id: %s, uuid: %s, fullname: %s, shortcode: %s' % (self.id, self.uuid,
-    name: str
+                self.fullname, self.shortcode)
-    shortcode: str
-    uuid: str = dataclasses.field(default_factory=_uuidgen)
 
 # TODO: This should go in the eventual romdb class.
 def get_file_sha1sum(filename):
@@ -50,34 +49,6 @@ def get_file_sha1sum(filename):
 
     return sha1sum.hexdigest()
 
-def _build_sql_constraints(inclusive, constraints):
-    '''Build an SQL constraint clause out of a dictionary.
-
-    inclusive   - If True, uses SQL AND operator, otherwise OR will be used.
-    constraints - A dictionary of arbitrary constraints
-
-    returns a tuple containing the appropriately formatted SQL string and a list of parameters
-    '''
-    if constraints == {}:
-        return ('', [])
-
-    if inclusive:
-        logical_separator = SQL_AND
-    else:
-        logical_separator = SQL_OR
-
-    sql_constraint_string = 'WHERE '
-    sql_parameter_list = []
-    for key, value in constraints.items():
-        sql_constraint_string += '%s=?%s' % (key, logical_separator)
-        sql_parameter_list.append(value)
-
-    # Trim off the last ', '
-    sql_constraint_string = sql_constraint_string[0:-len(logical_separator)]
-
-    return (sql_constraint_string, sql_parameter_list)
-
-
 class MetadataDB:
     def __init__(self, db_path):
         '''
@@ -85,21 +56,14 @@ class MetadataDB:
         Either way, create a connection object.
         '''
         # TODO: This process needs real error handling.
-        self._connection = sqlite3.connect(db_path)
+        # TODO: Add DAT import/credit support.
-
+        self._engine = sqlalchemy.create_engine('sqlite:///%s' % db_path)
-        with self._connection:
-            self._connection.execute('CREATE TABLE IF NOT EXISTS images (uuid PRIMARY KEY, '
-                                     'sha1sum UNIQUE NOT NULL, filename NOT NULL, release_group);') 
-
-            self._connection.execute('CREATE TABLE IF NOT EXISTS release_groups (UUID PRIMARY KEY, '
-                                     'name NOT NULL, platform NOT NULL);')
-
-            self._connection.execute('CREATE TABLE IF NOT EXISTS platforms (UUID PRIMARY KEY, '
-                                     'name NOT NULL, shortcode UNIQUE NOT NULL);')
-
-            # TODO: Add DAT import support.
 
-            print('Database initialized.')
+        _SQLBase.metadata.create_all(self._engine)
+        _db_session_maker.configure(bind=self._engine)
+        # TODO: Using One Big Session may have unintended consequences in other, less linear
+        # applications. For now, it works.
+        self._session = _db_session_maker()
 
     def add_image(self, image_data):
         pass
@@ -119,38 +83,32 @@ class MetadataDB:
     def remove_release_group(self, release_group_data):
         pass
 
-    def add_platform(self, platform_data):
+    def add_platform(self, platform):
         ''' Add a platform shortcode to the database. '''
-        values = list(dataclasses.asdict(platform_data).values())
+        self._session.add(platform)
-        print(values)
+        self._session.commit()
-        with self._connection:
-            self._connection.execute('INSERT INTO platforms VALUES (?, ?, ?);', values)
 
-    def update_platform(self, platform_data):
+    def update_platform(self, platform):
         pass
 
-    # TODO: Switch this to use more generic constraints language
+    def remove_platform(self, platform):
-    def remove_platform(self, platform_shortcode):
+        '''Remove a specific platform from the database. '''
-        with self._connection:
+        self._session.delete(platform)
-            self._connection.execute('DELETE FROM platforms WHERE shortcode=?;', platform_shortcode)
+        self._session.commit()
 
     def search_platforms(self, inclusive=True, **constraints):
         '''Search for platforms, given the parameters. '''
 
-        sql_where_clause, sql_parameters = _build_sql_constraints(inclusive, constraints)
+        query = self._session.query(Platform)
 
-        platform_data_list = []
+        for key, value in constraints.items():
-        with self._connection:
+            query = query.filter(getattr(Platform, key).ilike('%%%s%%' % value))
-            cursor = self._connection.cursor()
-            sql_query = 'SELECT * FROM platforms %s;' % sql_where_clause
-            cursor.execute(sql_query, sql_parameters)
-            rows = cursor.fetchall()
 
-            for row in rows:
+        platform_list = []
-                platform_data = PlatformData(*row)
+        for platform in query:
-                platform_data_list.append(platform_data)
+            platform_list.append(platform)
 
-        return platform_data_list
+        return platform_list
 
     def add_dat(self, dat_data):
         pass

lark

@@ -80,6 +80,14 @@ for filename in os.listdir(search_dir):
 SMD_DAT_PATH = '/home/lumia/Downloads/Sega - Mega Drive - Genesis (20200303-035539).dat'
 TEST_HASH = 'cfbf98c36c776677290a872547ac47c53d2761d6'
 
+def _kwargs_parse(kwargs_list):
+    kwargs = {}
+    for kwarg_string in kwargs_list:
+        key, value = kwarg_string.split('=')
+        kwargs[key] = value
+
+    return kwargs
+
 action_object = sys.argv[1]
 action = sys.argv[2]
 
@@ -92,7 +100,7 @@ if action_object == 'platform':
         platform_shortcode = sys.argv[3]
         platform_name = sys.argv[4]
 
-        platform_data = hashdb.PlatformData(shortcode=platform_shortcode,
+        platform_data = hashdb.Platform(shortcode=platform_shortcode,
                                             name=platform_name)
 
         print(platform_data)
@@ -100,27 +108,28 @@ if action_object == 'platform':
         db.add_platform(platform_data)
 
     elif action == 'list':
-        # TODO: convert this into a dict before passing.
+        # TODO: Filter support is exclusively limited to SQLAlchemy's filter.ilike function. Figure
-        # TODO: Abstract out constraint parsing etc.
+        # out a good way to include other filters.
-        constraints = sys.argv[3:]
+        filters = _kwargs_parse(sys.argv[3:])
-        platform_results = db.search_platforms()
+        platform_results = db.search_platforms(**filters)
         print(platform_results)
 
     elif action == 'remove':
         constraints = sys.argv[3:]
-        db.remove_platform(constraints)
+        platforms = db.search_platforms(constraints)
+        for platform in platforms:
+            print('Removing %s.' % platform.fullname)
+            db.remove_platform(platform)
 
     elif action == 'test':
         # TODO: Delete this action before merging into dev. It's just for ugly testing.
         platform_shortcode = sys.argv[3]
         platform_name = sys.argv[4]
 
-        platform_data = hashdb.PlatformData(shortcode=platform_shortcode,
+        platform_data = hashdb.Platform(shortcode=platform_shortcode,
-                                            name=platform_name)
+                                            fullname=platform_name)
-
-        import dataclasses
-        print(dataclasses.asdict(platform_data).values())
-
 
+        #db.add_platform(platform_data)
+        print(db.search_platforms())
 else:
     print('Unknown object.')