Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support creating unique top level keys for KKV data store #613

Merged
merged 11 commits into from
Nov 13, 2023
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
from typing import Any

from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.kkv_data_store import KKVDataStoreModel
from spiffworkflow_backend.models.script_attributes_context import ScriptAttributesContext
from spiffworkflow_backend.scripts.script import Script


class CreateUniqueKKVTopLevelKey(Script):
@staticmethod
def requires_privileged_permissions() -> bool:
"""We have deemed this function safe to run without elevated permissions."""
return False

def get_description(self) -> str:
return (
"Creates a new unique KKV top level key using the provided prefix. "
"This allows for a safe construction of incrementing keys, such as study1, study2."
)

def run(self, script_attributes_context: ScriptAttributesContext, *args: Any, **kwargs: Any) -> Any:
top_level_key_prefix = args[0]

model = KKVDataStoreModel(top_level_key="", secondary_key="", value={})
db.session.add(model)
db.session.commit()

top_level_key = f"{top_level_key_prefix}{model.id}"

db.session.delete(model)
db.session.commit()

return top_level_key
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
from collections.abc import Generator

import pytest
from flask.app import Flask
from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.kkv_data_store import KKVDataStoreModel
from spiffworkflow_backend.scripts.create_unique_kkv_top_level_key import CreateUniqueKKVTopLevelKey

from tests.spiffworkflow_backend.helpers.base_test import BaseTest


@pytest.fixture()
def with_clean_data_store(app: Flask, with_db_and_bpmn_file_cleanup: None) -> Generator[None, None, None]:
db.session.query(KKVDataStoreModel).delete()
db.session.commit()
yield


class TestCreateUniqueKKVTopLevelKey(BaseTest):
"""Infer from class name."""

# TODO: skipped due to sqlite handling of rowid - https://www.sqlite.org/autoinc.html
# since the immediate use case does not require sqlite will defer the fix until needed
def skip_test_creates_unique_top_level_keys(self, with_clean_data_store: None) -> None:
prefix = "study_"
ids = [
CreateUniqueKKVTopLevelKey().run(None, prefix), # type: ignore
CreateUniqueKKVTopLevelKey().run(None, prefix), # type: ignore
CreateUniqueKKVTopLevelKey().run(None, prefix), # type: ignore
CreateUniqueKKVTopLevelKey().run(None, prefix), # type: ignore
CreateUniqueKKVTopLevelKey().run(None, prefix), # type: ignore
CreateUniqueKKVTopLevelKey().run(None, prefix), # type: ignore
CreateUniqueKKVTopLevelKey().run(None, prefix), # type: ignore
]
unique_ids = set(ids)
assert len(ids) == len(unique_ids)
5 changes: 2 additions & 3 deletions spiffworkflow-frontend/package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading