diff --git a/stix2/datastore/relational_db/database_backends/sqlite_backend.py b/stix2/datastore/relational_db/database_backends/sqlite_backend.py new file mode 100644 index 00000000..545f928b --- /dev/null +++ b/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 diff --git a/stix2/datastore/relational_db/relational_db_testing.py b/stix2/datastore/relational_db/relational_db_testing.py index 54bf0aae..acd0727a 100644 --- a/stix2/datastore/relational_db/relational_db_testing.py +++ b/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,