Initial changes to support SQLite backend.

stix2/datastore/relational_db/database_backends/sqlite_backend.py

@@ -0,0 +1,51 @@
+import os
+from typing import Any
+
+from sqlalchemy import TIMESTAMP, LargeBinary, Text
+from sqlalchemy import event
+
+from stix2.base import (
+    _DomainObject, _MetaObject, _Observable, _RelationshipObject,
+)
+from stix2.datastore.relational_db.utils import schema_for
+
+from .database_backend_base import DatabaseBackend
+
+
+class SQLiteBackend(DatabaseBackend):
+    default_database_connection_url = f"sqlite:///stix-data-sink.db"
+
+    def __init__(self, database_connection_url=default_database_connection_url, force_recreate=False, **kwargs: Any):
+        super().__init__(database_connection_url, force_recreate=force_recreate, **kwargs)
+
+        @event.listens_for(self.database_connection, "connect")
+        def set_sqlite_pragma(dbapi_connection, connection_record):
+            cursor = dbapi_connection.cursor()
+            cursor.execute("PRAGMA foreign_keys=ON")
+            result = cursor.execute("PRAGMA foreign_keys")
+            for row in result:
+                print('PRAGMA foreign_keys:', row)
+            cursor.close()
+
+    # =========================================================================
+    # sql type methods (overrides)
+
+    @staticmethod
+    def determine_sql_type_for_binary_property():  # noqa: F811
+        return SQLiteBackend.determine_sql_type_for_string_property()
+
+    @staticmethod
+    def determine_sql_type_for_hex_property():  # noqa: F811
+        # return LargeBinary
+        return SQLiteBackend.determine_sql_type_for_string_property()
+
+    @staticmethod
+    def determine_sql_type_for_timestamp_property():  # noqa: F811
+        return TIMESTAMP(timezone=True)
+
+    # =========================================================================
+    # Other methods
+
+    @staticmethod
+    def array_allowed():
+        return False

stix2/datastore/relational_db/relational_db_testing.py

@@ -1,6 +1,7 @@
 import datetime as dt
 
 from database_backends.postgres_backend import PostgresBackend
+from database_backends.sqlite_backend import SQLiteBackend
 import pytz
 
 import stix2
@@ -287,7 +288,8 @@ def test_dictionary():
 
 def main():
     store = RelationalDBStore(
-        PostgresBackend("postgresql://localhost/stix-data-sink", force_recreate=True),
+        #PostgresBackend("postgresql://localhost/stix-data-sink", force_recreate=True),
+        SQLiteBackend("sqlite:///stix-data-sink.db", force_recreate=True),
         True,
         None,
         True,