From 8b0ce6fda7f3a51241ccd9dfd09ee0169a8410e6 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Tue, 23 Jun 2026 12:51:48 +0000 Subject: [PATCH 01/15] fix: fall back to diskcache when memcached is inaccessible Wraps the pylibmc client in _MemcachedWithFallback so that any connection error on get/set/delete/flush_all is caught and silently retried against the local diskcache instance. The server is still probed at startup and an error is logged, but the application no longer crashes when memcached is temporarily unreachable. Co-Authored-By: Claude Sonnet 4.6 --- hydra_base/lib/cache.py | 51 +++++++++++++++++++++++++++++++++++++---- 1 file changed, 46 insertions(+), 5 deletions(-) diff --git a/hydra_base/lib/cache.py b/hydra_base/lib/cache.py index 713ae036..d916256e 100644 --- a/hydra_base/lib/cache.py +++ b/hydra_base/lib/cache.py @@ -17,27 +17,68 @@ def _init_diskcache(): global cache import diskcache as dc cache = dc.Cache(tempfile.gettempdir()) + +class _MemcachedWithFallback: + """Wraps a pylibmc client and falls back to diskcache on connection errors.""" + + def __init__(self, pylibmc_cache, fallback): + self._mc = pylibmc_cache + self._fb = fallback + + def set(self, key, value, *args, **kwargs): + try: + return self._mc.set(key, value, *args, **kwargs) + except Exception as e: + log.warning("Memcached set failed (%s), falling back to diskcache.", e) + return self._fb.set(key, value) + + def get(self, key, *args, **kwargs): + try: + return self._mc.get(key, *args, **kwargs) + except Exception as e: + log.warning("Memcached get failed (%s), falling back to diskcache.", e) + return self._fb.get(key) + + def delete(self, key, *args, **kwargs): + try: + return self._mc.delete(key, *args, **kwargs) + except Exception as e: + log.warning("Memcached delete failed (%s), falling back to diskcache.", e) + return self._fb.delete(key, retry=False) + + def flush_all(self): + try: + self._mc.flush_all() + except Exception as e: + log.warning("Memcached flush_all failed (%s), falling back to diskcache.", e) + self._fb.clear() + + if hydraconfig.get('cache', 'type') != "memcached": _init_diskcache() elif hydraconfig.get('cache', 'type') == 'memcached': try: import pylibmc + import diskcache as dc + host = hydraconfig.get('cache', 'host', '127.0.0.1') port = hydraconfig.get('cache', 'port', 11211) - cache = pylibmc.Client([f"{host}:{port}"], binary=True) + _mc = pylibmc.Client([f"{host}:{port}"], binary=True) # Check if Memcached server is reachable by setting a test key test_key = "__connection_test__" - #pick a unique key based on the time test_value = datetime.datetime.toordinal(datetime.datetime.now()) try: - cache.set(test_key, test_value, 1) - cache.get(test_key) + _mc.set(test_key, test_value, 1) + _mc.get(test_key) log.info("Connected to memcached server.") except Exception: raise ConnectionError("Memcached server not responding.") + _fallback = dc.Cache(tempfile.gettempdir()) + cache = _MemcachedWithFallback(_mc, _fallback) + except (ModuleNotFoundError, ConnectionError) as e: if isinstance(e, ModuleNotFoundError): log.warning("Unable to find pylibmc. Defaulting to diskcache.") @@ -48,6 +89,6 @@ def _init_diskcache(): def clear_cache(): if hasattr(cache, 'flush_all'): - cache.flush_all() # memcache + cache.flush_all() # memcache / wrapped memcache else: cache.clear() # diskcache From 249fec679e0d8fbf33bb26181e257c4f7da7a206 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Tue, 23 Jun 2026 12:53:08 +0000 Subject: [PATCH 02/15] refactor: remove zope.sqlalchemy dependency Replace zope.sqlalchemy register() / transaction.commit() / transaction.abort() with plain SQLAlchemy session calls (DBSession.commit() / DBSession.rollback()). Remove the zope.sqlalchemy package from requirements.txt. Co-Authored-By: Claude Sonnet 4.6 --- hydra_base/db/__init__.py | 8 ++------ hydra_base/util/storage.py | 6 ++---- requirements.txt | 1 - 3 files changed, 4 insertions(+), 11 deletions(-) diff --git a/hydra_base/db/__init__.py b/hydra_base/db/__init__.py index 57119e59..6e9a7e14 100644 --- a/hydra_base/db/__init__.py +++ b/hydra_base/db/__init__.py @@ -26,11 +26,8 @@ from sqlalchemy.engine import Engine from .. import config -from zope.sqlalchemy import register - from hydra_base.exceptions import HydraError -import transaction from sqlalchemy.orm import sessionmaker, declarative_base import logging @@ -157,7 +154,6 @@ def connect(db_url=None): maker = sessionmaker(bind=engine, autoflush=False, autocommit=False) DBSession = scoped_session(maker) - register(DBSession) global DeclarativeBase try: @@ -173,10 +169,10 @@ def get_session(): def commit_transaction(): try: - transaction.commit() + DBSession.commit() except Exception as e: log.critical(e) - transaction.abort() + DBSession.rollback() def open_session(): log.debug("OPENING SESSION") diff --git a/hydra_base/util/storage.py b/hydra_base/util/storage.py index 25abd98a..1c026ce2 100644 --- a/hydra_base/util/storage.py +++ b/hydra_base/util/storage.py @@ -11,8 +11,6 @@ import json import logging import os -import transaction - from bson.objectid import ObjectId import pymongo from pymongo import MongoClient @@ -200,7 +198,7 @@ def export_dataset_to_external_storage(ds_id: int, db_name: str=None, collection external_token = mongo_config["direct_location_token"] md = Metadata(key=location_key, value=external_token) dataset.metadata.append(md) - transaction.commit() + db.DBSession.commit() return result @@ -243,7 +241,7 @@ def import_dataset_from_external_storage(ds_id: int, db_name: str=None, collecti if m.key == location_key: break dataset.metadata.pop(idx) - transaction.commit() + db.DBSession.commit() result = path.delete_one({"_id": object_id}) if result.deleted_count != 1: diff --git a/requirements.txt b/requirements.txt index 61844b8f..206d3010 100644 --- a/requirements.txt +++ b/requirements.txt @@ -23,4 +23,3 @@ requests s3fs sqlalchemy tables -zope.sqlalchemy From 114e737a8f72ea0093edeb01fe762ad95f14a3e4 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Tue, 23 Jun 2026 12:53:29 +0000 Subject: [PATCH 03/15] fix: correct rule and template management functions - Rule.__json__: expose format, ref_key, network_id, template_id and types so callers receive a complete representation - Template.check_write_permission: new helper that gates writes on admin status (imported _is_admin from base) - get_network_rules: access template_id through rtype.templatetype rather than directly on rtype; fix get_hierarchy().reverse() (which returned None) by calling reverse() on the list separately; pass user_id to get_hierarchy() - update_rule: assign ref_id to rule_i.template_id (not network_id) for TEMPLATE ref_key - get_template: validate that a cached template has templatetypes before returning it; re-fetch from DB if the cached entry is empty Co-Authored-By: Claude Sonnet 4.6 --- hydra_base/db/model/rule.py | 7 +- hydra_base/db/model/template.py | 9 + hydra_base/lib/rules.py | 9 +- hydra_base/lib/template/__init__.py | 7 +- tests/templates/test_template_cache.py | 265 +++++++++++++++++++++++++ 5 files changed, 290 insertions(+), 7 deletions(-) create mode 100644 tests/templates/test_template_cache.py diff --git a/hydra_base/db/model/rule.py b/hydra_base/db/model/rule.py index e3211595..997065e5 100644 --- a/hydra_base/db/model/rule.py +++ b/hydra_base/db/model/rule.py @@ -176,9 +176,14 @@ def asdict(self): "id": self.id, "name": self.name, "value": self.value, + "format": self.format, + "ref_key": self.ref_key, + "network_id": self.network_id, + "template_id": self.template_id, "description": self.description, "status": self.status, - "owners": self.owners + "owners": self.owners, + "types": [{"code": t.code} for t in self.types] } diff --git a/hydra_base/db/model/template.py b/hydra_base/db/model/template.py index 69316cbc..69deb0b7 100644 --- a/hydra_base/db/model/template.py +++ b/hydra_base/db/model/template.py @@ -17,6 +17,7 @@ # along with HydraPlatform. If not, see # from .base import * +from .base import _is_admin __all__ = ['Template', 'TemplateType', 'TypeAttr', 'ResourceType'] @@ -323,6 +324,14 @@ def get_hierarchy(self, user_id): hierarchy = hierarchy + self.parent.get_hierarchy(user_id) return hierarchy + def check_write_permission(self, user_id, do_raise=True): + if _is_admin(user_id): + return True + if do_raise: + raise PermissionError("Permission denied. User %s does not have edit" + " access on template %s" % (user_id, self.id)) + return False + class TemplateType(Base, Inspect): """ Template Type diff --git a/hydra_base/lib/rules.py b/hydra_base/lib/rules.py index 4098b337..e253fc66 100644 --- a/hydra_base/lib/rules.py +++ b/hydra_base/lib/rules.py @@ -66,11 +66,12 @@ def get_network_rules(network_id, summary=True, **kwargs): #all rules associated to them. all_template_rules = [] for rtype in network.types: - if not hasattr(rtype, "template_id"): + if not hasattr(rtype, 'templatetype') or not hasattr(rtype.templatetype, "template_id"): continue - template = db.DBSession.query(Template).filter(Template.id==rtype.template_id).one() + template = db.DBSession.query(Template).filter(Template.id==rtype.templatetype.template_id).one() #need this to go top-bottom to apply rules from the top level down - template_hierarchy = template.get_hierarchy().reverse() + template_hierarchy = template.get_hierarchy(user_id) + template_hierarchy.reverse() for current_template in template_hierarchy: this_template_rules = rule_qry.filter(Rule.template_id == current_template.id).all() all_template_rules = all_template_rules + this_template_rules @@ -279,7 +280,7 @@ def update_rule(rule, **kwargs): elif rule.ref_key.upper() == 'PROJECT': rule_i.network_id = rule.project_id if rule.project_id else rule.ref_id elif rule.ref_key.upper() == 'TEMPLATE': - rule_i.network_id = rule.template_id if rule.template_id else rule.ref_id + rule_i.template_id = rule.template_id if rule.template_id else rule.ref_id else: raise HydraError("Ref Key {0} not recognised.".format(rule.ref_key)) diff --git a/hydra_base/lib/template/__init__.py b/hydra_base/lib/template/__init__.py index a6755b37..90e8ebdc 100644 --- a/hydra_base/lib/template/__init__.py +++ b/hydra_base/lib/template/__init__.py @@ -743,8 +743,11 @@ def get_template(template_id, **kwargs): tmpl_j = cache.get(f"{CACHE_KEY}_{template_id}") if tmpl_j is not None: - log.info("Returning cached template") - return JSONObject(tmpl_j) + result = JSONObject(tmpl_j) if not isinstance(tmpl_j, JSONObject) else tmpl_j + if result.templatetypes is not None: + log.info("Returning cached template") + return result + log.warning("Cached template %s has no templatetypes — re-fetching from DB", template_id) try: log.info("Building template") diff --git a/tests/templates/test_template_cache.py b/tests/templates/test_template_cache.py new file mode 100644 index 00000000..9e200ef2 --- /dev/null +++ b/tests/templates/test_template_cache.py @@ -0,0 +1,265 @@ +# -*- coding: utf-8 -*- + +""" +Tests that demonstrate the template cache-poisoning bug. + +Root cause: _save_template_to_cache() is called from add_template() and +import_template_dict() using JSONObject(orm_object) directly, without first +calling get_types(). This produces a cached entry where 'templatetypes' is +either absent or contains raw ORM-derived data rather than the fully-resolved +list that get_types() produces. + +When get_template() subsequently hits that cache entry it returns a JSONObject +whose .templatetypes is None (via JSONObject.__getattr__'s default of None for +missing keys). Any caller that then iterates over .templatetypes — most notably +complexmodels.Template.__init__ in hydra-server — raises: + + TypeError: 'NoneType' object is not iterable + +The three tests below each isolate a different aspect of the bug: + +1. test_poisoned_cache_entry_yields_none_templatetypes + Directly poisons the cache (simulating what add_template/import_template_dict + do) and asserts that get_template returns a template with templatetypes=None, + which then raises TypeError on iteration. + +2. test_add_template_poisons_cache + Calls the real add_template() and confirms it writes a cache entry that is + missing or has fewer types than get_types() would return, so that a fresh + get_template() call on the same id returns incorrect data. + +3. test_import_template_dict_poisons_cache + Same as above but via the import_template_dict() code path, which has the + identical _save_template_to_cache call without get_types(). +""" + +import datetime +import json +import tempfile +import time +import pytest + +import hydra_base as hb +from hydra_base.lib.objects import JSONObject +from hydra_base.lib.cache import cache, clear_cache +from hydra_base.lib import template as template_lib +from hydra_base.lib import attributes as attr_lib +from hydra_base.lib.template import CACHE_KEY +from hydra_base.util.hdb import ( + create_default_users_and_perms, + make_root_user, + create_default_units_and_dimensions, +) + + +# --------------------------------------------------------------------------- +# Session-scoped DB fixture (self-contained SQLite, no conftest.py dependency) +# --------------------------------------------------------------------------- + +@pytest.fixture(scope='module') +def db(): + millis = int(round(time.time() * 1000)) + db_url = f'sqlite:///{tempfile.gettempdir()}/test_template_cache_{millis}.db' + hb.db.connect(db_url) + create_default_users_and_perms() + make_root_user() + create_default_units_and_dimensions() + yield hb.db + clear_cache() + hb.db.close_session() + + +@pytest.fixture(autouse=True) +def _clear_template_cache(): + """Wipe template cache entries before each test for isolation.""" + yield + clear_cache() + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + +USER_ID = 1 # root user created by make_root_user() + + +def _create_attr(name=None): + if name is None: + name = f"cache_test_attr_{datetime.datetime.now().timestamp()}" + attr = JSONObject() + attr.name = name + attr.dimension_id = None + return attr_lib.add_attribute(attr, user_id=USER_ID) + + +def _make_template_input(attr_id, name=None): + """Return a JSONObject suitable for template_lib.add_template().""" + if name is None: + name = f"CacheBugTest {datetime.datetime.now()}" + tmpl = JSONObject() + tmpl.name = name + tmpl.templatetypes = [] + + t = JSONObject() + t.name = "NodeType" + t.resource_type = 'NODE' + t.typeattrs = [JSONObject({'attr_id': attr_id})] + tmpl.templatetypes.append(t) + return tmpl + + +def _cached_entry(template_id): + return cache.get(f"{CACHE_KEY}_{template_id}") + + +# --------------------------------------------------------------------------- +# tests +# --------------------------------------------------------------------------- + +class TestTemplateCachePoisoning: + + def test_poisoned_cache_entry_is_bypassed(self, db): + """ + Verifies the protection in get_template: when the cache contains an + entry with templatetypes absent (the poisoned state produced by + add_template / import_template_dict), get_template detects this and + falls back to the DB path rather than returning None and causing: + + TypeError: 'NoneType' object is not iterable (complexmodels.py:739) + """ + attr = _create_attr() + new_tmpl = template_lib.add_template( + _make_template_input(attr.id), user_id=USER_ID + ) + template_id = new_tmpl.id + + # Poison the cache with an entry that has no 'templatetypes' key — + # this is what add_template / import_template_dict can write when the + # ORM relationship is not present in __dict__ at caching time. + poisoned = JSONObject({'id': template_id, 'name': new_tmpl.name}) + assert 'templatetypes' not in poisoned + cache.set(f"{CACHE_KEY}_{template_id}", poisoned) + + # The fix: get_template detects the missing key and re-fetches from DB. + result = template_lib.get_template(template_id, user_id=USER_ID) + + # Protection holds: templatetypes is not None even though the cache was poisoned + assert result.templatetypes is not None, ( + "get_template should fall back to DB when the cached entry has no " + "templatetypes — the protection in get_template is not working." + ) + + # Iteration must not raise TypeError (the production crash is prevented) + types = list(result.templatetypes) + assert len(types) == 1 + + def test_add_template_poisons_cache(self, db): + """ + add_template() calls _save_template_to_cache(JSONObject(tmpl)) without + calling get_types() first (lib/template/__init__.py lines 521-522). + + The cached entry may not contain the same 'templatetypes' data that a + fresh get_template() DB query would produce via get_types(). After + add_template the cache is populated; a subsequent get_template() hits + that entry and should return properly-populated templatetypes — but + the bug means it may not. + """ + attr = _create_attr() + tmpl_input = _make_template_input(attr.id) + expected_type_count = len(tmpl_input.templatetypes) + + new_tmpl = template_lib.add_template(tmpl_input, user_id=USER_ID) + template_id = new_tmpl.id + + # Confirm add_template DID write a cache entry + cached = _cached_entry(template_id) + assert cached is not None, ( + "add_template should populate the cache via _save_template_to_cache" + ) + + # get_template hits the cache entry written by add_template + # (not the fully-resolved get_types() path) + result = JSONObject(template_lib.get_template(template_id, user_id=USER_ID)) + + # Bug: templatetypes may be absent (None) from the cached JSONObject + # because _save_template_to_cache was called without get_types() + assert result.templatetypes is not None, ( + "templatetypes is None after get_template — the cache entry written " + "by add_template is missing templatetypes because get_types() was " + "not called before _save_template_to_cache (lines 521-522)" + ) + + actual_count = len(list(result.templatetypes)) + assert actual_count == expected_type_count, ( + f"Expected {expected_type_count} type(s) but got {actual_count}. " + "The cache entry from add_template differs from what get_types() returns." + ) + + # Each type must have typeattrs — get_types() fetches these explicitly, + # but JSONObject(orm_obj) only includes them if the relationship was loaded. + for tt in result.templatetypes: + assert tt.typeattrs is not None, ( + f"Type '{tt.name}' has typeattrs=None in the cached entry. " + "get_types() loads typeattrs explicitly; the raw ORM-based " + "JSONObject may not." + ) + + def test_import_template_dict_poisons_cache(self, db): + """ + import_template_dict() has the identical bug: it calls + _save_template_to_cache(JSONObject(template_i)) without get_types() + (lib/template/__init__.py lines 459-460). + + After import, get_template hits the poisoned entry and may return a + template where templatetypes is None or typeattrs is missing. + """ + attr = _create_attr() + + template_dict = { + "attributes": { + str(attr.id): {"id": attr.id, "name": attr.name} + }, + "datasets": {}, + "template": { + "name": f"ImportCacheBugTest {datetime.datetime.now()}", + "templatetypes": [ + { + "name": "ImportedNodeType", + "resource_type": "NODE", + "typeattrs": [{"attr_id": attr.id}] + } + ] + } + } + + imported = template_lib.import_template_dict( + template_dict, allow_update=True, user_id=USER_ID + ) + template_id = imported.id + + # Confirm import_template_dict DID write a cache entry (lines 459-460) + cached = _cached_entry(template_id) + assert cached is not None, ( + "import_template_dict should populate the cache via " + "_save_template_to_cache (lines 459-460)" + ) + + # get_template hits the cache entry written by import_template_dict + result = JSONObject(template_lib.get_template(template_id, user_id=USER_ID)) + + assert result.templatetypes is not None, ( + "templatetypes is None after get_template following " + "import_template_dict — the cache was poisoned because " + "_save_template_to_cache was called without get_types() " + "(lib/template/__init__.py lines 459-460)" + ) + + types = list(result.templatetypes) + assert len(types) == 1 + + for tt in types: + assert tt.typeattrs is not None, ( + f"Type '{tt.name}' has typeattrs=None — import_template_dict " + "caches without calling get_types(), so typeattrs may not be " + "present in the cached JSONObject." + ) From 6eb213f2e243f47a5e8a6010fe0c6f6195a3979f Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Tue, 23 Jun 2026 12:55:10 +0000 Subject: [PATCH 04/15] perf: speed up bulk resource-attribute and scenario-data insertion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resource attributes (add_resource_attributes): - Replace sequential exists-check + individual ORM inserts with a single bulk_insert_ignore() call (new helper in db/__init__ using dialect- specific INSERT IGNORE / ON CONFLICT DO NOTHING). - Query back inserted IDs in one batch using tuple_() IN instead of iterating each object. - Invalidate the network RA cache immediately after the bulk insert. Scenario data (bulk_update_resourcedata / assign_value): - Pre-compute all dataset hashes in a single pass before entering the per-RS loop and batch-check for hash collisions with one query. - Skip the DB write entirely for unchanged datasets (hash match). - In the single-RS fast lane, mutate the dataset in place and consult the pre-built collision cache instead of issuing a per-dataset query. - Cache newly created datasets by hash within the batch to avoid duplicate inserts when the same value appears more than once. - Defer flush() to the end of the batch (flush=False per iteration). - Eager-load dataset metadata with joinedload to eliminate N+1 queries. Other: - Add thread-local permission cache in check_perm() to avoid repeated identical DB queries within the same request. - Remove unused user_id/source parameters from get_metadata_as_dict(). - Update test assertions to match the new return shape of add_resource_attributes (dict of id → [resource_id, attr_id]). Co-Authored-By: Claude Sonnet 4.6 --- hydra_base/db/__init__.py | 33 ++++- hydra_base/lib/attributes.py | 67 +++++----- hydra_base/lib/objects.py | 6 +- hydra_base/lib/scenario.py | 183 +++++++++++++++++++++------- hydra_base/util/permissions.py | 9 ++ tests/attributes/test_attributes.py | 9 +- 6 files changed, 217 insertions(+), 90 deletions(-) diff --git a/hydra_base/db/__init__.py b/hydra_base/db/__init__.py index 6e9a7e14..daabcc7b 100644 --- a/hydra_base/db/__init__.py +++ b/hydra_base/db/__init__.py @@ -194,8 +194,37 @@ def close_session(): def rollback_transaction(): - #import pudb; pudb.set_trace() - transaction.abort() + DBSession.rollback() + +def bulk_insert_ignore(model, rows): + """ + Bulk insert rows into model, silently skipping any that would violate a + unique constraint. Cross-database compatible. + + Does not return inserted IDs — query back as needed after calling. + """ + if not rows: + return + + if engine is None: + raise HydraError("bulk_insert_ignore: No database engine available. Please call connect() first.") + + dialect_name = engine.dialect.name + + if dialect_name == 'mysql': + from sqlalchemy.dialects.mysql import insert as _insert + stmt = _insert(model).values(rows).prefix_with('IGNORE') + elif dialect_name == 'postgresql': + from sqlalchemy.dialects.postgresql import insert as _insert + stmt = _insert(model).values(rows).on_conflict_do_nothing() + elif dialect_name == 'sqlite': + from sqlalchemy.dialects.sqlite import insert as _insert + stmt = _insert(model).values(rows).on_conflict_do_nothing() + else: + raise HydraError(f"bulk_insert_ignore: unsupported dialect '{dialect_name}'") + + DBSession.execute(stmt) + def restart_session(caller='-- not specified --'): """ diff --git a/hydra_base/lib/attributes.py b/hydra_base/lib/attributes.py index c7b5a730..a5515029 100644 --- a/hydra_base/lib/attributes.py +++ b/hydra_base/lib/attributes.py @@ -23,12 +23,10 @@ from collections import defaultdict -from sqlalchemy import or_, and_, func +from sqlalchemy import or_, and_, func, tuple_ from sqlalchemy.orm import aliased, joinedload from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.exc import IntegrityError -from zope.sqlalchemy import mark_changed - from ..db.model import Attr,\ User,\ Node,\ @@ -96,8 +94,10 @@ def _get_resource(ref_key, ref_id): raise ResourceNotFoundError("Resource %s with ID %s not found"%(ref_key, ref_id)) def _get_resource_id(ra): - if ra.resource_id is not None: + try: return ra.resource_id + except AttributeError: + pass ref_key = ra.ref_key if ref_key == 'NETWORK': @@ -866,17 +866,9 @@ def add_resource_attributes(resource_attributes, **kwargs): if len(resource_attributes) == 0: return {} - #1. Identify the network ID + #1. Identify the network ID (needed for cache invalidation) network_id = get_network_id_from_resource_attribute(resource_attributes[0]) - #2. Get all the resource attributes in the network - network_resource_attributes = get_all_network_resourceattributes(network_id, **kwargs) - #3. Remove any duplicates from the incoming data in case there are RAs which are already there - network_ra_lookup = {(ra.attr_id, ra.ref_key, _get_resource_id(ra)): ra for ra in network_resource_attributes} - - #an RA in the database has a 'REF_KEY' column, and then a 'network_id', 'node_id', 'link_id', 'group_id' and 'project_id' column which are mutually exclusive. - #The incoming RA can have this format, but also 'ref_key', and 'ref_id', where ref_id is the ID of the resource, and ref_key is the type of resource. - #The incoming RA can also have a 'resource_type' and 'resource_id' column, which is the same as the ref_key and ref_id. - #The result should be a ref_key and the relevant resource_id column (network_id, node_id etc) set to the ID of the resource. + key_to_field = { 'NETWORK': 'network_id', 'NODE': 'node_id', @@ -919,34 +911,31 @@ def add_resource_attributes(resource_attributes, **kwargs): if target_field: ra[target_field] = ra['ref_id'] - ras_to_be_inserted = [] + #2. Build DB-column-filtered dicts for all incoming RAs + cols = set(c.name for c in ResourceAttr.__table__.columns) - {'id', 'cr_date', 'updated_at'} + rows = [{k: v for k, v in ra.items() if k in cols} for ra in resource_attributes] - for ra in resource_attributes: - key = (ra.attr_id, ra['ref_key'], _get_resource_id(ra)) - if key not in network_ra_lookup: - ras_to_be_inserted.append(ra) + #3. Bulk insert — the DB skips any rows that violate a unique constraint + if rows: + log.info("Inserting %s resource attributes (duplicates will be skipped)", len(rows)) + db.bulk_insert_ignore(ResourceAttr, rows) + db.DBSession.flush() + cache.delete(f'network_resource_attributes_{network_id}') + #4. Query back IDs for all requested (attr_id, resource_id) combinations inserted_ids = {} - - #4. Add the new resource attributes - cols = list(filter(lambda x: x not in ['id', 'cr_date', 'updated_at'], [c.name for c in ResourceAttr.__table__.columns])) - ras_to_be_inserted = [{k: v for k, v in ra.items() if k in cols} for ra in ras_to_be_inserted] - - if len(ras_to_be_inserted) > 0: - log.info("Adding %s new resource attributes", len(ras_to_be_inserted)) - objs = [ResourceAttr(**ra) for ra in ras_to_be_inserted] - db.DBSession.add_all(objs) - db.DBSession.flush() # or commit - for obj in objs: - inserted_ids[(obj.get_resource_id(), obj.attr_id)] = obj.id - # Mark the session as dirty to ensure that the changes are saved - # This is necessary if you are using a session with autocommit=False - mark_changed(db.DBSession()) - - cache.set(f'network_resource_attributes_{network_id}', - cache.get(f'network_resource_attributes_{network_id}') + [JSONObject(obj) for obj in objs], 60*60) - - db.DBSession.flush() + by_ref_key = defaultdict(list) + for row in rows: + by_ref_key[row['ref_key']].append(row) + + for ref_key, ras in by_ref_key.items(): + id_field = key_to_field[ref_key] + id_col = getattr(ResourceAttr, id_field) + pairs = [(ra['attr_id'], ra[id_field]) for ra in ras] + for obj in db.DBSession.query(ResourceAttr).filter( + tuple_(ResourceAttr.attr_id, id_col).in_(pairs) + ).all(): + inserted_ids[obj.id] = [_get_resource_id(obj), obj.attr_id] return inserted_ids diff --git a/hydra_base/lib/objects.py b/hydra_base/lib/objects.py index 36568a16..e92b682a 100644 --- a/hydra_base/lib/objects.py +++ b/hydra_base/lib/objects.py @@ -339,14 +339,10 @@ def parse_value(self): log.exception(e) raise HydraError("Error parsing value %s: %s"%(self.value, e)) - def get_metadata_as_dict(self, user_id=None, source=None): + def get_metadata_as_dict(self): """ Convert a metadata json string into a dictionary. - Args: - user_id (int): Optional: Insert user_id into the metadata if specified - source (string): Optional: Insert source (the name of the app typically) into the metadata if necessary. - Returns: dict: THe metadata as a python dictionary """ diff --git a/hydra_base/lib/scenario.py b/hydra_base/lib/scenario.py index c92e4a92..31a049e6 100644 --- a/hydra_base/lib/scenario.py +++ b/hydra_base/lib/scenario.py @@ -971,8 +971,6 @@ def bulk_update_resourcedata(scenario_ids, resource_scenarios, **kwargs): Update the data associated with a list of scenarios. """ user_id = kwargs.get('user_id') - res = None - res = {} net_ids = db.DBSession.query(Scenario.network_id).filter(Scenario.id.in_(scenario_ids)).all() @@ -991,17 +989,86 @@ def bulk_update_resourcedata(scenario_ids, resource_scenarios, **kwargs): #ones that have been passed in to avoid querying for every one individually. ra_ids = [rs.resource_attr_id for rs in resource_scenarios] - r_scens_i = db.DBSession.query(ResourceScenario).filter( + r_scens_i = db.DBSession.query(ResourceScenario)\ + .options(joinedload(ResourceScenario.dataset).joinedload(Dataset.metadata))\ + .filter( ResourceScenario.scenario_id == scenario_id, ResourceScenario.resource_attr_id.in_(ra_ids)).all() r_scen_dict = dict((rs.resource_attr_id, rs) for rs in r_scens_i) + + existing_dataset_ids = [r.dataset_id for r in r_scens_i if r.dataset_id] + if existing_dataset_ids: + _rows = db.DBSession.query( + ResourceScenario.dataset_id, + ResourceScenario.scenario_id, + ResourceScenario.resource_attr_id + ).filter(ResourceScenario.dataset_id.in_(existing_dataset_ids)).all() + dataset_rs_map = {} + for row in _rows: + dataset_rs_map.setdefault(row.dataset_id, []).append((row.scenario_id, row.resource_attr_id)) + else: + dataset_rs_map = {} + + new_dataset_cache = {} + + # Pre-pass: identify update-in-place candidates, compute their new hashes, + # then do ONE batch collision check against existing DB datasets. + # This replaces per-dataset hash-collision queries in the main loop. + # + # Hash is computed from the raw value string (same as Dataset.set_hash uses + # value_ref), NOT from parse_value() output. parse_value() returns Python + # objects whose str() representation differs from the raw JSON for non-scalar + # types (e.g. str([[1, 2]]) == '[[1, 2]]' but value_ref stores '[[1,2]]'). + _prepass = {} # resource_attr_id -> (current_dataset_id, new_hash) + _unchanged_ra_ids = set() # RS where hash matched — skip in main loop + for rs_in in resource_scenarios: + if rs_in.dataset is None: + continue + r_scen = r_scen_dict.get(rs_in.resource_attr_id) + if r_scen is None or r_scen.dataset_id is None: + continue + ds_j = JSONDataset(rs_in.dataset) + raw_val = str(ds_j.value) if ds_j.value is not None else None + if raw_val is None or raw_val.upper().strip() in ('NULL', ''): + continue + meta = ds_j.get_metadata_as_dict() + new_hash = ds_j.get_hash(raw_val, meta) + connected = dataset_rs_map.get(r_scen.dataset_id, []) + if r_scen.dataset.hash == new_hash: + _unchanged_ra_ids.add(rs_in.resource_attr_id) + elif len(connected) == 1 and connected[0][0] == scenario_id and connected[0][1] == rs_in.resource_attr_id: + _prepass[rs_in.resource_attr_id] = (r_scen.dataset_id, new_hash) + + if _prepass: + _cand_hashes = list({h for _, h in _prepass.values()}) + _cand_ids = list({did for did, _ in _prepass.values()}) + _collision_rows = db.DBSession.query(Dataset).filter( + Dataset.hash.in_(_cand_hashes), + Dataset.id.notin_(_cand_ids) + ).all() + # Pre-seed with DB-existing collisions; updated entries are added during the loop. + dataset_hash_cache = {row.hash: row for row in _collision_rows} + else: + dataset_hash_cache = {} + for rs in resource_scenarios: if rs.dataset is not None: + ra_id = rs.resource_attr_id + if ra_id in _unchanged_ra_ids: + # Pre-pass confirmed hash match — no DB write needed. + r_scen_i = r_scen_dict.get(ra_id) + if r_scen_i is not None: + res[str(scenario_id)].append(r_scen_i) + continue updated_rs = _update_resourcescenario(scen_i, rs, - r_scen_i=r_scen_dict.get(rs.resource_attr_id), + r_scen_i=r_scen_dict.get(ra_id), user_id=user_id, - source=kwargs.get('app_name')) + source=kwargs.get('app_name'), + flush=False, + dataset_rs_map=dataset_rs_map, + new_dataset_cache=new_dataset_cache, + dataset_hash_cache=dataset_hash_cache) #this is cast as a string so it can be read into a JSONObject res[str(scenario_id)].append(updated_rs) else: @@ -1121,7 +1188,7 @@ def _delete_resourcescenario(scenario_id, resource_attr_id, suppress_error=False db.DBSession.delete(sd_i) db.DBSession.flush() -def _update_resourcescenario(scenario, resource_scenario, r_scen_i=None, dataset=None, new=False, user_id=None, source=None): +def _update_resourcescenario(scenario, resource_scenario, r_scen_i=None, dataset=None, new=False, user_id=None, source=None, flush=True, dataset_rs_map=None, new_dataset_cache=None, dataset_hash_cache=None): """ Insert or Update the value of a resource's attribute by first getting the resource, then parsing the input data, then assigning the value. @@ -1164,8 +1231,13 @@ def _update_resourcescenario(scenario, resource_scenario, r_scen_i=None, dataset dataset = resource_scenario.dataset dataset_j = JSONDataset(dataset) - value = dataset_j.parse_value() + metadata = dataset_j.get_metadata_as_dict() + data_unit_id = dataset_j.unit_id + # Use raw string value for hash — matches how Dataset.set_hash() uses value_ref. + # parse_value() returns Python objects whose str() representation differs from + # the raw JSON for non-scalar types, causing false "changed" detections. + data_hash = dataset_j.get_hash(str(dataset_j.value), metadata) log.debug("Assigning %s to resource attribute: %s", value, ra_id) @@ -1173,11 +1245,6 @@ def _update_resourcescenario(scenario, resource_scenario, r_scen_i=None, dataset log.info("Cannot set data on resource attribute %s", ra_id) return None - metadata = dataset_j.get_metadata_as_dict(source=source, user_id=user_id) - data_unit_id = dataset_j.unit_id - - data_hash = dataset_j.get_hash(value, metadata) - new_rscen_i = assign_value(r_scen_i, dataset_j.type.lower(), value, @@ -1186,13 +1253,19 @@ def _update_resourcescenario(scenario, resource_scenario, r_scen_i=None, dataset metadata=metadata, data_hash=data_hash, user_id=user_id, - source=source) + source=source, + flush=flush, + dataset_rs_map=dataset_rs_map, + new_dataset_cache=new_dataset_cache, + dataset_hash_cache=dataset_hash_cache) return new_rscen_i @required_perms("edit_data", "edit_network") def assign_value(rs, data_type, val, - unit_id, name, metadata={}, data_hash=None, user_id=None, source=None): + unit_id, name, metadata={}, data_hash=None, user_id=None, source=None, + flush=True, dataset_rs_map=None, new_dataset_cache=None, + dataset_hash_cache=None): """ Insert or update a piece of data in a scenario. If the dataset is being shared by other resource scenarios, a new dataset is inserted. @@ -1215,45 +1288,74 @@ def assign_value(rs, data_type, val, #Has this dataset changed? if rs.dataset.hash == data_hash: - log.info("Dataset has not changed. Returning.") + log.debug("Dataset has not changed. Returning.") return rs - connected_rs = db.DBSession.query(ResourceScenario).filter(ResourceScenario.dataset_id == rs.dataset.id).all() - #If there's no RS found, then the incoming rs is new, so the dataset can be altered - #without fear of affecting something else. - if len(connected_rs) == 0: - #If it's 1, the RS exists in the DB, but it's the only one using this dataset or - #The RS isn't in the DB yet and the datset is being used by 1 other RS. - update_dataset = True + if dataset_rs_map is not None: + connected = dataset_rs_map.get(rs.dataset.id, []) + else: + connected = [(r.scenario_id, r.resource_attr_id) + for r in db.DBSession.query(ResourceScenario).filter( + ResourceScenario.dataset_id == rs.dataset.id).all()] - if len(connected_rs) == 1: - if connected_rs[0].scenario_id == rs.scenario_id and connected_rs[0].resource_attr_id == rs.resource_attr_id: + if len(connected) == 1: + if connected[0][0] == rs.scenario_id and connected[0][1] == rs.resource_attr_id: update_dataset = True else: update_dataset = False if update_dataset is True: - log.info("Updating dataset '%s'", name) - dataset = data.update_dataset(rs.dataset.id, name, data_type, val, unit_id, metadata, flush=False, **dict(user_id=user_id)) - log.info("Updated dataset '%s'", name) + log.debug("Updating dataset '%s'", name) + if dataset_rs_map is not None: + # Bulk-path fast lane: we already know this dataset has exactly one RS + # (this one) and the scenario is unlocked (checked above), so we can + # skip the resourcescenarios lazy-load and the scenario.locked traversal. + dataset = rs.dataset + dataset.type = data_type + dataset.value = val + dataset.set_metadata(metadata) + dataset.unit_id = unit_id + dataset.name = name + dataset.created_by = user_id + new_hash = dataset.set_hash() + + # Collision check using the pre-built batch cache — no per-dataset DB query. + # dataset_hash_cache is pre-seeded with DB-existing datasets that share a + # candidate hash, and is updated during the loop to catch batch-internal + # collisions (two update-path datasets converging on the same hash). + if dataset_hash_cache is not None: + existing = dataset_hash_cache.get(new_hash) + if existing is not None and existing.id != dataset.id and existing.check_read_permission(user_id, do_raise=False): + db.DBSession.delete(dataset) + dataset = existing + else: + dataset_hash_cache[new_hash] = dataset + else: + dataset = data.update_dataset(rs.dataset.id, name, data_type, val, unit_id, metadata, flush=False, **dict(user_id=user_id)) + log.debug("Updated dataset '%s'", name) rs.dataset = dataset rs.dataset_id = dataset.id - log.info("Set RS dataset id to %s"%dataset.id) + log.debug("Set RS dataset id to %s"%dataset.id) else: - log.info("Creating new dataset %s in scenario %s", name, rs.scenario_id) - dataset = data.add_dataset( - data_type, - val, - unit_id, - metadata=metadata, - name=name, - **dict(user_id=user_id) - ) + log.debug("Creating new dataset %s in scenario %s", name, rs.scenario_id) + if new_dataset_cache is not None and data_hash in new_dataset_cache: + dataset = new_dataset_cache[data_hash] + else: + dataset = data.add_dataset( + data_type, + val, + unit_id, + metadata=metadata, + name=name, + **dict(user_id=user_id) + ) + if new_dataset_cache is not None and data_hash is not None: + new_dataset_cache[data_hash] = dataset rs.dataset = dataset rs.source = source - - db.DBSession.flush() + if flush: + db.DBSession.flush() return rs @@ -1283,8 +1385,7 @@ def add_data_to_attribute(scenario_id, resource_attr_id, dataset,**kwargs): dataset_j = JSONDataset(dataset) value = dataset_j.parse_value() - dataset_metadata = dataset_j.get_metadata_as_dict(user_id=kwargs.get('user_id'), - source=kwargs.get('source')) + dataset_metadata = dataset_j.get_metadata_as_dict() if value is None: raise HydraError(f"Cannot set value to attribute. No value was sent with dataset {dataset_j.id}") diff --git a/hydra_base/util/permissions.py b/hydra_base/util/permissions.py index 18be1b27..db839603 100644 --- a/hydra_base/util/permissions.py +++ b/hydra_base/util/permissions.py @@ -17,12 +17,14 @@ # along with HydraPlatform. If not, see # +import threading from functools import wraps from .. import db from ..db.model import Perm, User, Role, RolePerm, RoleUser from sqlalchemy.orm.exc import NoResultFound from ..exceptions import PermissionError +_perm_cache = threading.local() def check_perm(user_id, permission_code): @@ -33,6 +35,11 @@ def check_perm(user_id, permission_code): If the user does not have permission to perfom an action, a permission error is thrown. """ + cache = _perm_cache.__dict__.setdefault('cache', {}) + key = (user_id, permission_code) + if key in cache: + return + try: perm = db.DBSession.query(Perm).filter(Perm.code==permission_code).one() except NoResultFound: @@ -51,6 +58,8 @@ def check_perm(user_id, permission_code): raise PermissionError("Permission denied. User %s does not have permission %s"% (user_id, permission_code)) + cache[key] = True + def check_role(user_id, role_code): """ Checks whether a user has been assigned the specified role diff --git a/tests/attributes/test_attributes.py b/tests/attributes/test_attributes.py index 581fc219..edcc9635 100644 --- a/tests/attributes/test_attributes.py +++ b/tests/attributes/test_attributes.py @@ -269,7 +269,9 @@ def test_add_resource_attributes(self, existing_attr = network_with_data.attributes[0] - #add one new one, plus one existing one. This should result in only one being added + # add one new one, plus one existing one. add_resource_attributes returns + # IDs for all requested RAs (new and pre-existing), so len(added_attrs) == 2, + # but the network only grows by 1. newattributes = [ {"attr_id": new_attr.id, "network_id": network_with_data.id, "attr_is_var": "Y"}, existing_attr @@ -279,8 +281,9 @@ def test_add_resource_attributes(self, updated_network = client.get_network(network_with_data.id) - assert (network_with_data.id, new_attr.id) in added_attrs - assert len(updated_network.attributes) == len(network_with_data.attributes) + len(added_attrs) + assert [network_with_data.id, new_attr.id] in list(added_attrs.values()) + assert len(added_attrs) == len(newattributes) + assert len(updated_network.attributes) == len(network_with_data.attributes) + 1 assert new_attr.id in [netattr.attr_id for netattr in updated_network.attributes] From 56ac61bdb58502fd8195c8438ee67673367f36d5 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Tue, 23 Jun 2026 12:55:34 +0000 Subject: [PATCH 05/15] feat: add DB migrations for cloned_network_id and cloned_project_id columns Adds nullable FK columns tNetwork.cloned_network_id and tProject.cloned_project_id so cloned networks and projects can track which original they were derived from. Co-Authored-By: Claude Sonnet 4.6 --- .../d4e9b1f2c83a_cloned_network_id.py | 38 +++++++++++++++++++ .../e7a3c9f04b12_cloned_project_id.py | 38 +++++++++++++++++++ 2 files changed, 76 insertions(+) create mode 100644 hydra_base/db/alembic/versions/d4e9b1f2c83a_cloned_network_id.py create mode 100644 hydra_base/db/alembic/versions/e7a3c9f04b12_cloned_project_id.py diff --git a/hydra_base/db/alembic/versions/d4e9b1f2c83a_cloned_network_id.py b/hydra_base/db/alembic/versions/d4e9b1f2c83a_cloned_network_id.py new file mode 100644 index 00000000..8a362804 --- /dev/null +++ b/hydra_base/db/alembic/versions/d4e9b1f2c83a_cloned_network_id.py @@ -0,0 +1,38 @@ +"""cloned_network_id + +Revision ID: d4e9b1f2c83a +Revises: a81a860cda39 +Create Date: 2026-05-20 00:00:00.000000 + +""" +import logging +from alembic import op +import sqlalchemy as sa + +log = logging.getLogger(__name__) + +# revision identifiers, used by Alembic. +revision = 'd4e9b1f2c83a' +down_revision = 'a81a860cda39' +branch_labels = None +depends_on = None + + +def upgrade(): + if op.get_bind().dialect.name == 'mysql': + try: + op.add_column('tNetwork', + sa.Column('cloned_network_id', + sa.Integer(), + sa.ForeignKey('tNetwork.id'), + nullable=True)) + except Exception as e: + log.critical(e) + + +def downgrade(): + if op.get_bind().dialect.name == 'mysql': + try: + op.drop_column('tNetwork', 'cloned_network_id') + except Exception as e: + log.critical(e) diff --git a/hydra_base/db/alembic/versions/e7a3c9f04b12_cloned_project_id.py b/hydra_base/db/alembic/versions/e7a3c9f04b12_cloned_project_id.py new file mode 100644 index 00000000..d5602350 --- /dev/null +++ b/hydra_base/db/alembic/versions/e7a3c9f04b12_cloned_project_id.py @@ -0,0 +1,38 @@ +"""cloned_project_id + +Revision ID: e7a3c9f04b12 +Revises: d4e9b1f2c83a +Create Date: 2026-05-20 00:00:00.000000 + +""" +import logging +from alembic import op +import sqlalchemy as sa + +log = logging.getLogger(__name__) + +# revision identifiers, used by Alembic. +revision = 'e7a3c9f04b12' +down_revision = 'd4e9b1f2c83a' +branch_labels = None +depends_on = None + + +def upgrade(): + if op.get_bind().dialect.name == 'mysql': + try: + op.add_column('tProject', + sa.Column('cloned_project_id', + sa.Integer(), + sa.ForeignKey('tProject.id'), + nullable=True)) + except Exception as e: + log.critical(e) + + +def downgrade(): + if op.get_bind().dialect.name == 'mysql': + try: + op.drop_column('tProject', 'cloned_project_id') + except Exception as e: + log.critical(e) From f98319d7d46a3cfcff6336fcd8125ede981fcb47 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Tue, 23 Jun 2026 16:06:53 +0000 Subject: [PATCH 06/15] fix diverged alembic heads --- .../a81a860cda39_attribute_scoping.py | 2 +- .../edf7bffb7b33_merge_divergent_heads.py | 24 +++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) create mode 100644 hydra_base/db/alembic/versions/edf7bffb7b33_merge_divergent_heads.py diff --git a/hydra_base/db/alembic/versions/a81a860cda39_attribute_scoping.py b/hydra_base/db/alembic/versions/a81a860cda39_attribute_scoping.py index 2c4ce215..001e8790 100644 --- a/hydra_base/db/alembic/versions/a81a860cda39_attribute_scoping.py +++ b/hydra_base/db/alembic/versions/a81a860cda39_attribute_scoping.py @@ -11,7 +11,7 @@ # revision identifiers, used by Alembic. revision = 'a81a860cda39' -down_revision = '04e4ae80b7b9' +down_revision = 'cec2b77ad85e' branch_labels = None depends_on = None diff --git a/hydra_base/db/alembic/versions/edf7bffb7b33_merge_divergent_heads.py b/hydra_base/db/alembic/versions/edf7bffb7b33_merge_divergent_heads.py new file mode 100644 index 00000000..26e7df49 --- /dev/null +++ b/hydra_base/db/alembic/versions/edf7bffb7b33_merge_divergent_heads.py @@ -0,0 +1,24 @@ +"""merge divergent heads + +Revision ID: edf7bffb7b33 +Revises: 580425ade2e4, 877adf863b33, a1b2c3d4e5f6 +Create Date: 2026-06-23 16:06:15.787461 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'edf7bffb7b33' +down_revision = ('580425ade2e4', '877adf863b33', 'a1b2c3d4e5f6') +branch_labels = None +depends_on = None + + +def upgrade(): + pass + + +def downgrade(): + pass From e92deb08a1c04132dc157ae0da4af6c8bb5c4368 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Wed, 24 Jun 2026 12:30:28 +0000 Subject: [PATCH 07/15] fix: Improve timing of the get_all_attributes_in_network by separating out groups which were causing bad DB query performance --- hydra_base/lib/network.py | 52 +++++++++++++++++++++++++++------------ 1 file changed, 36 insertions(+), 16 deletions(-) diff --git a/hydra_base/lib/network.py b/hydra_base/lib/network.py index 07072656..01f73eb1 100644 --- a/hydra_base/lib/network.py +++ b/hydra_base/lib/network.py @@ -2681,22 +2681,42 @@ def get_all_attributes_in_network(network_id, **kwargs): raise HydraError("Network %s not found" % (network_id,)) net.check_read_permission(user_id) - network_attr_ids = db.DBSession.query(ResourceAttr.attr_id.label('attr_id')).filter( - ResourceAttr.network_id == network_id - ) - node_attr_ids = db.DBSession.query(ResourceAttr.attr_id.label('attr_id')).join( - Node, ResourceAttr.node_id == Node.id - ).filter(Node.network_id == network_id) - link_attr_ids = db.DBSession.query(ResourceAttr.attr_id.label('attr_id')).join( - Link, ResourceAttr.link_id == Link.id - ).filter(Link.network_id == network_id) - group_attr_ids = db.DBSession.query(ResourceAttr.attr_id.label('attr_id')).join( - ResourceGroup, ResourceAttr.group_id == ResourceGroup.id - ).filter(ResourceGroup.network_id == network_id) - - all_attr_ids = network_attr_ids.union(node_attr_ids, link_attr_ids, group_attr_ids).subquery() - - attrs = db.DBSession.query(Attr).join(all_attr_ids, Attr.id == all_attr_ids.c.attr_id).all() + #Find the distinct attr_ids used by the network and its nodes/links/groups. + attr_ids = set() + + #Network/Node/Link lookups join through an indexed network_id and are fast. + for r in db.DBSession.query(ResourceAttr.attr_id).filter( + ResourceAttr.network_id == network_id).distinct().all(): + attr_ids.add(r[0]) + + for r in db.DBSession.query(ResourceAttr.attr_id).join( + Node, ResourceAttr.node_id == Node.id).filter( + Node.network_id == network_id).distinct().all(): + attr_ids.add(r[0]) + + for r in db.DBSession.query(ResourceAttr.attr_id).join( + Link, ResourceAttr.link_id == Link.id).filter( + Link.network_id == network_id).distinct().all(): + attr_ids.add(r[0]) + + #The ResourceGroup join must NOT be done in one query: tResourceGroup has no + #dedicated index on network_id (only a composite unique index), so MySQL + #mis-plans the join and scans the whole multi-million-row tResourceAttr + #table - ~12s for a single group. Fetch the (tiny) set of group ids first, + #then look up ResourceAttr by group_id, which uses the (group_id, attr_id) + #covering index directly. + group_ids = [r[0] for r in db.DBSession.query(ResourceGroup.id).filter( + ResourceGroup.network_id == network_id).all()] + + if len(group_ids) > 0: + for r in db.DBSession.query(ResourceAttr.attr_id).filter( + ResourceAttr.group_id.in_(group_ids)).distinct().all(): + attr_ids.add(r[0]) + + if len(attr_ids) == 0: + return [] + + attrs = db.DBSession.query(Attr).filter(Attr.id.in_(attr_ids)).all() return [JSONObject(a) for a in attrs] From 8f20a75e039c8fb4405bc7198312f1ff94f795d2 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Wed, 8 Jul 2026 14:22:06 +0000 Subject: [PATCH 08/15] Update the get_projects folder to include any networks contained in nav projects --- hydra_base/lib/project.py | 42 ++++++++++++++++++--- tests/project/test_project_inheritance.py | 46 +++++++++++++++++++++++ 2 files changed, 83 insertions(+), 5 deletions(-) diff --git a/hydra_base/lib/project.py b/hydra_base/lib/project.py index fcd19c31..d5f6459b 100644 --- a/hydra_base/lib/project.py +++ b/hydra_base/lib/project.py @@ -378,17 +378,49 @@ def get_projects(uid, include_shared_projects=True, projects_ids_list_filter=Non #to projects further down the tree which they are owners of. nav_project_ids = set(Project.get_cache(uid).get(project_id, [])) - scoped_project_ids nav_projects_i = db.DBSession.query(Project).filter(Project.id.in_(nav_project_ids)).filter(Project.parent_id==project_id).all() + + user = db.DBSession.query(User).filter(User.id == req_user_id).one() + isadmin = user.is_admin() + + #nav_project_ids aren't in projects_i (the user doesn't own/have direct + #view access to them), so their networks aren't in project_network_lookup + #below -- look them up separately, otherwise a nav_only project always + #shows "no networks" even when the user can actually see some of its + #networks (the rest of its contents may only be visible via a + #sub-project further down, which get_projects_networks doesn't reach). + nav_network_lookup = get_projects_networks( + [p.id for p in nav_projects_i], uid, isadmin=isadmin, **kwargs) + + #Also account for networks sitting one level further down, in a nav + #project's own direct sub-projects -- Project.get_cache(uid) already + #has the full tree in memory so this is just a dict lookup per nav + #project, plus one extra batched network query (not a recursive walk). + nav_child_project_ids = {} + all_nav_child_ids = [] + for nav_project_i in nav_projects_i: + children = [ + cid for cid in Project.get_cache(uid).get(nav_project_i.id, []) + if cid != nav_project_i.id + ] + nav_child_project_ids[nav_project_i.id] = children + all_nav_child_ids.extend(children) + + nav_child_network_lookup = get_projects_networks( + all_nav_child_ids, uid, isadmin=isadmin, **kwargs) if all_nav_child_ids else {} + nav_projects = [] for nav_project_i in nav_projects_i: nav_project_j = JSONObject(nav_project_i) nav_project_j.nav_only = True nav_project_j.owners = [] - nav_project_j.networks = [] - nav_projects.append(nav_project_j) + direct_networks = nav_network_lookup.get(nav_project_i.id, []) + child_ids = nav_child_project_ids[nav_project_i.id] + child_networks = [n for cid in child_ids for n in nav_child_network_lookup.get(cid, [])] - user = db.DBSession.query(User).filter(User.id == req_user_id).one() - isadmin = user.is_admin() + nav_project_j.networks = direct_networks + child_networks + + nav_projects.append(nav_project_j) project_network_lookup = get_projects_networks([p.id for p in projects_i], uid, isadmin=isadmin, **kwargs) @@ -452,7 +484,7 @@ def get_projects_networks(project_ids, uid, isadmin=None, **kwargs): .filter(Network.project_id.in_(project_ids),\ Network.status=='A') if not isadmin: - network_qry.outerjoin(NetworkOwner)\ + network_qry = network_qry.outerjoin(NetworkOwner)\ .filter( NetworkOwner.user_id == uid, NetworkOwner.view == 'Y' diff --git a/tests/project/test_project_inheritance.py b/tests/project/test_project_inheritance.py index 233cf068..6e1050d6 100644 --- a/tests/project/test_project_inheritance.py +++ b/tests/project/test_project_inheritance.py @@ -318,6 +318,14 @@ def test_access_to_shared_network_in_sub_project(self, client, projectmaker, net userc_projects = client.get_projects(pytest.user_c.id) assert proj1.id in [p.id for p in userc_projects] + #proj1 is only visible to User C as a nav-only project (they don't + #own it or have direct view access) and it doesn't directly contain + #net1 -- net1 is one level down, in proj2. Regression check: proj1's + #entry should still list net1, otherwise proj1 looks like it has "no + #networks" even though the user can see one further down the tree. + userc_proj1_entry = next(p for p in userc_projects if p.id == proj1.id) + assert net1.id in [n.id for n in userc_proj1_entry.networks] + #User C doesn't have explicit read access on proj1 or proj2, but should #be abe to navigate to proj1 and 2 so they cna access proj4 userc_proj1 = client.get_project(project_id=proj1.id) @@ -349,6 +357,44 @@ def test_access_to_shared_network_in_sub_project(self, client, projectmaker, net with pytest.raises(HydraError): client.get_project(project_id=proj3.id) + #client is module-scoped -- leaving user_id set to User C here would + #silently corrupt every subsequent test in this module (it did, + #before this line was added: their projectmaker/networkmaker calls + #would run as User C instead of the intended owner). + client.user_id = proj_user + + def test_get_projects_networks_permission_filtering(self, client, projectmaker, networkmaker): + """ + Regression test for a bug in get_projects_networks() where the + non-admin permission filter (an outerjoin onto NetworkOwner) was + applied to a new query object without reassigning it back to + network_qry, making the filter a silent no-op. This meant a + non-admin user calling get_projects() could see every network in + a project they can navigate to, including ones never shared with + them individually. + """ + client.user_id = 1 # force current user to be 1 to avoid potential inconsistencies + proj_user = client.user_id + proj = projectmaker.create(share=False) + net1 = networkmaker.create(project_id=proj.id) + net2 = networkmaker.create(project_id=proj.id) + + #Share only net1 with User C. This is enough to give them nav-only + #visibility of the containing project, but NOT of net2. + client.share_network(net1.id, ['UserC'], False, False) + + client.user_id = pytest.user_c.id + userc_projects = client.get_projects(pytest.user_c.id) + userc_proj_entry = next(p for p in userc_projects if p.id == proj.id) + + userc_network_ids = {n.id for n in userc_proj_entry.networks} + assert userc_network_ids == {net1.id} + assert net2.id not in userc_network_ids + + client.user_id = proj_user + + client.user_id = proj_user + def test_remove_project_parent(self, client, projectmaker, networkmaker): """ Test two actions which should result in a project's parent_id From e3e00dd02d84035c5667e4fa212e3f4f40c3133e Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Wed, 8 Jul 2026 14:22:41 +0000 Subject: [PATCH 09/15] Wrap dataset in StringIO when reading json --- hydra_base/util/dataset_util.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/hydra_base/util/dataset_util.py b/hydra_base/util/dataset_util.py index 3d9439fb..64a14204 100644 --- a/hydra_base/util/dataset_util.py +++ b/hydra_base/util/dataset_util.py @@ -19,6 +19,7 @@ import logging from decimal import Decimal +from io import StringIO from operator import mul from ..exceptions import HydraError, ValidationError @@ -44,7 +45,7 @@ def json_to_df(json_dataframe): data_dict = json.loads(json_dataframe) #load the json dataframe into a pandas dataframe - df = pd.read_json(json_dataframe, convert_axes=False) + df = pd.read_json(StringIO(json_dataframe), convert_axes=False) #extraxt the ordered index from the dict ordered_index = list(data_dict[df.columns[0]].keys()) From 11a9885d83466af60bffca435b0a8addb3685596 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Thu, 9 Jul 2026 14:44:31 +0000 Subject: [PATCH 10/15] Add TTL to check_perm's per-thread permission cache The cache previously stored positive permission checks forever with no invalidation, so revoking a user's role/permission wouldn't take effect for a long-lived worker thread until it happened to restart. Bound this by re-verifying against the DB after PERM_CACHE_TTL seconds (default 30, configurable via [permissions] cache_ttl). --- hydra_base/util/permissions.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/hydra_base/util/permissions.py b/hydra_base/util/permissions.py index db839603..ce628cde 100644 --- a/hydra_base/util/permissions.py +++ b/hydra_base/util/permissions.py @@ -18,14 +18,20 @@ # import threading +import time from functools import wraps -from .. import db +from .. import db, config from ..db.model import Perm, User, Role, RolePerm, RoleUser from sqlalchemy.orm.exc import NoResultFound from ..exceptions import PermissionError _perm_cache = threading.local() +#Seconds a positive permission check is cached for before being re-verified +#against the DB. Keeps check_perm fast for hot paths while bounding how long +#a revoked permission/role can remain (incorrectly) usable. +PERM_CACHE_TTL = config.getint('permissions', 'cache_ttl', 30) + def check_perm(user_id, permission_code): """ @@ -37,7 +43,8 @@ def check_perm(user_id, permission_code): """ cache = _perm_cache.__dict__.setdefault('cache', {}) key = (user_id, permission_code) - if key in cache: + cached_at = cache.get(key) + if cached_at is not None and time.time() - cached_at < PERM_CACHE_TTL: return try: @@ -58,7 +65,7 @@ def check_perm(user_id, permission_code): raise PermissionError("Permission denied. User %s does not have permission %s"% (user_id, permission_code)) - cache[key] = True + cache[key] = time.time() def check_role(user_id, role_code): """ From 5c4197365df6a66eb52c75988e853dfee6976d4b Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Thu, 9 Jul 2026 14:44:42 +0000 Subject: [PATCH 11/15] Harden bulk_update_resourcedata's batch fast lane Bundle the three batch-scoped caches (dataset_rs_map, new_dataset_cache, dataset_hash_cache) into a single _BulkAssignContext, document the connectivity invariant it relies on, and re-verify single-ownership at the point of the in-place dataset mutation rather than trusting an earlier check to still hold. Add a per-batch summary log line (unchanged / updated in place / created / collisions avoided). Also fixes a real bug surfaced by the new tests: when a hash collision was found but the calling user had no read permission on the existing dataset, the code fell through and tried to persist a second dataset with the same hash, which always raised IntegrityError against tDataset.hash's UNIQUE constraint (both in this fast lane and in the pre-existing data.add_dataset). Fixed via a new Dataset.set_unique_hash() that salts the hash computation only, without leaking the salt into the dataset's real (persisted) metadata. --- hydra_base/db/model/dataset.py | 23 +++++ hydra_base/lib/data.py | 7 +- hydra_base/lib/scenario.py | 127 ++++++++++++++++++----- tests/test_scenario.py | 180 +++++++++++++++++++++++++++++++++ 4 files changed, 311 insertions(+), 26 deletions(-) diff --git a/hydra_base/db/model/dataset.py b/hydra_base/db/model/dataset.py index 0585682e..1447eb57 100644 --- a/hydra_base/db/model/dataset.py +++ b/hydra_base/db/model/dataset.py @@ -16,6 +16,8 @@ # You should have received a copy of the GNU Lesser General Public License # along with HydraPlatform. If not, see # +import uuid + from .base import * from hydra_base.lib.storage import ( @@ -160,6 +162,27 @@ def set_hash(self,metadata=None): return data_hash + def set_unique_hash(self, metadata=None): + """ + Like set_hash(), but guarantees the result cannot collide with any + other dataset's hash. Used when a hash collision was found but the + existing dataset can't be reused (e.g. no read permission on it) -- + tDataset.hash has a DB-level UNIQUE constraint, so leaving the hash + as a duplicate would raise IntegrityError on flush. + + The salt is folded into the hash computation only -- it is NOT + passed to set_metadata()/persisted as a real metadata row, since + that would leak an internal implementation detail into the + dataset's actual (user-visible) metadata. + """ + if metadata is None: + metadata = self.get_metadata_as_dict() + + salted_metadata = dict(metadata) + salted_metadata['_hash_salt'] = uuid.uuid4().hex + + return self.set_hash(metadata=salted_metadata) + def get_metadata_as_dict(self): metadata = {} sortedmeta = sorted(self.metadata, key=lambda x:x.key.lower()) diff --git a/hydra_base/lib/data.py b/hydra_base/lib/data.py index 4694744d..62495520 100644 --- a/hydra_base/lib/data.py +++ b/hydra_base/lib/data.py @@ -486,7 +486,12 @@ def add_dataset(data_type, val, unit_id=None, metadata={}, name="", user_id=None if existing_dataset.check_read_permission(user_id, do_raise=False) is True: d = existing_dataset else: - d.set_hash() + #Can't reuse the existing dataset (no read permission) and can't keep + #this hash either -- tDataset.hash has a UNIQUE constraint, so leaving + #it as-is would raise IntegrityError on flush. set_unique_hash() salts + #the hash computation only, it does not touch this dataset's real + #(persisted) metadata. + d.hash = d.set_unique_hash(metadata) db.DBSession.add(d) except NoResultFound: db.DBSession.add(d) diff --git a/hydra_base/lib/scenario.py b/hydra_base/lib/scenario.py index 31a049e6..aaba09a7 100644 --- a/hydra_base/lib/scenario.py +++ b/hydra_base/lib/scenario.py @@ -965,6 +965,37 @@ def get_dataset_scenarios(dataset_id, **kwargs): return scenarios + +class _BulkAssignContext: + """ + Groups the batch-scoped caches that bulk_update_resourcedata builds once + per scenario_id and passes down through _update_resourcescenario and + assign_value, so those two functions can skip per-row DB round trips. + + INVARIANT: dataset_rs_map must contain *complete* connectivity info + (every (scenario_id, resource_attr_id) pair currently pointing at each + dataset_id) for every resource scenario being processed in this batch. + assign_value's in-place-mutation fast lane trusts this map to decide + whether a dataset is safe to mutate directly -- if the map is partial + or stale, that fast lane can silently corrupt a dataset that some + other, unlisted resource scenario still depends on. Build a fresh + instance per batch; never reuse one across a different set of + resource_scenarios. + """ + __slots__ = ('dataset_rs_map', 'new_dataset_cache', 'dataset_hash_cache', + 'unchanged', 'updated_in_place', 'created', 'collisions_avoided') + + def __init__(self, dataset_rs_map=None, new_dataset_cache=None, dataset_hash_cache=None): + self.dataset_rs_map = dataset_rs_map if dataset_rs_map is not None else {} + self.new_dataset_cache = new_dataset_cache if new_dataset_cache is not None else {} + self.dataset_hash_cache = dataset_hash_cache if dataset_hash_cache is not None else {} + #Counters purely for the end-of-batch summary log -- not used for any logic. + self.unchanged = 0 + self.updated_in_place = 0 + self.created = 0 + self.collisions_avoided = 0 + + @required_perms("edit_data", "edit_network") def bulk_update_resourcedata(scenario_ids, resource_scenarios, **kwargs): """ @@ -1009,8 +1040,6 @@ def bulk_update_resourcedata(scenario_ids, resource_scenarios, **kwargs): else: dataset_rs_map = {} - new_dataset_cache = {} - # Pre-pass: identify update-in-place candidates, compute their new hashes, # then do ONE batch collision check against existing DB datasets. # This replaces per-dataset hash-collision queries in the main loop. @@ -1051,6 +1080,9 @@ def bulk_update_resourcedata(scenario_ids, resource_scenarios, **kwargs): else: dataset_hash_cache = {} + bulk_ctx = _BulkAssignContext(dataset_rs_map=dataset_rs_map, + dataset_hash_cache=dataset_hash_cache) + for rs in resource_scenarios: if rs.dataset is not None: ra_id = rs.resource_attr_id @@ -1058,6 +1090,7 @@ def bulk_update_resourcedata(scenario_ids, resource_scenarios, **kwargs): # Pre-pass confirmed hash match — no DB write needed. r_scen_i = r_scen_dict.get(ra_id) if r_scen_i is not None: + bulk_ctx.unchanged += 1 res[str(scenario_id)].append(r_scen_i) continue updated_rs = _update_resourcescenario(scen_i, @@ -1066,9 +1099,7 @@ def bulk_update_resourcedata(scenario_ids, resource_scenarios, **kwargs): user_id=user_id, source=kwargs.get('app_name'), flush=False, - dataset_rs_map=dataset_rs_map, - new_dataset_cache=new_dataset_cache, - dataset_hash_cache=dataset_hash_cache) + bulk_ctx=bulk_ctx) #this is cast as a string so it can be read into a JSONObject res[str(scenario_id)].append(updated_rs) else: @@ -1076,6 +1107,12 @@ def bulk_update_resourcedata(scenario_ids, resource_scenarios, **kwargs): db.DBSession.flush() + log.info( + "bulk_update_resourcedata scenario %s: %s unchanged, %s updated in place, " + "%s created, %s collisions avoided (%s total)", + scenario_id, bulk_ctx.unchanged, bulk_ctx.updated_in_place, + bulk_ctx.created, bulk_ctx.collisions_avoided, len(resource_scenarios)) + return res @required_perms("edit_data", "edit_network") @@ -1188,11 +1225,15 @@ def _delete_resourcescenario(scenario_id, resource_attr_id, suppress_error=False db.DBSession.delete(sd_i) db.DBSession.flush() -def _update_resourcescenario(scenario, resource_scenario, r_scen_i=None, dataset=None, new=False, user_id=None, source=None, flush=True, dataset_rs_map=None, new_dataset_cache=None, dataset_hash_cache=None): +def _update_resourcescenario(scenario, resource_scenario, r_scen_i=None, dataset=None, new=False, user_id=None, source=None, flush=True, bulk_ctx=None): """ Insert or Update the value of a resource's attribute by first getting the resource, then parsing the input data, then assigning the value. + bulk_ctx (_BulkAssignContext): Optional. Set by bulk_update_resourcedata + to share its batch-scoped caches with assign_value. See + _BulkAssignContext's docstring for the invariant it requires. + returns a ResourceScenario object. """ if scenario is None: @@ -1255,22 +1296,30 @@ def _update_resourcescenario(scenario, resource_scenario, r_scen_i=None, dataset user_id=user_id, source=source, flush=flush, - dataset_rs_map=dataset_rs_map, - new_dataset_cache=new_dataset_cache, - dataset_hash_cache=dataset_hash_cache) + bulk_ctx=bulk_ctx) return new_rscen_i @required_perms("edit_data", "edit_network") def assign_value(rs, data_type, val, unit_id, name, metadata={}, data_hash=None, user_id=None, source=None, - flush=True, dataset_rs_map=None, new_dataset_cache=None, - dataset_hash_cache=None): + flush=True, bulk_ctx=None): """ Insert or update a piece of data in a scenario. If the dataset is being shared by other resource scenarios, a new dataset is inserted. If the dataset is ONLY being used by the resource scenario in question, the dataset is updated to avoid unnecessary duplication. + + bulk_ctx (_BulkAssignContext): Optional. When set (only by + bulk_update_resourcedata's batch path), dataset connectivity and + hash-collision lookups use bulk_ctx's pre-built caches instead of + a per-call DB query. bulk_ctx.dataset_rs_map MUST reflect complete + connectivity for every dataset touched in the batch -- see + _BulkAssignContext's docstring. The in-place mutation fast lane + below re-verifies single-ownership from that same map immediately + before mutating, so a caller-side bug that violates the invariant + fails loudly (HydraError) rather than silently corrupting a + dataset some other resource scenario still depends on. """ log.debug("Assigning value %s to rs %s in scenario %s", @@ -1291,8 +1340,8 @@ def assign_value(rs, data_type, val, log.debug("Dataset has not changed. Returning.") return rs - if dataset_rs_map is not None: - connected = dataset_rs_map.get(rs.dataset.id, []) + if bulk_ctx is not None: + connected = bulk_ctx.dataset_rs_map.get(rs.dataset.id, []) else: connected = [(r.scenario_id, r.resource_attr_id) for r in db.DBSession.query(ResourceScenario).filter( @@ -1306,10 +1355,26 @@ def assign_value(rs, data_type, val, if update_dataset is True: log.debug("Updating dataset '%s'", name) - if dataset_rs_map is not None: + if bulk_ctx is not None: # Bulk-path fast lane: we already know this dataset has exactly one RS # (this one) and the scenario is unlocked (checked above), so we can # skip the resourcescenarios lazy-load and the scenario.locked traversal. + # + # Defense in depth: re-verify single-ownership right here, at the point + # of the in-place mutation, rather than trusting the check a few lines + # above still holds. This doesn't add an independent data source (it's + # the same dataset_rs_map), but it means a future edit that moves code + # around, or a caller that passes an incomplete/stale map, fails loudly + # instead of silently mutating a dataset another resource scenario + # still depends on. + if not (len(connected) == 1 and connected[0][0] == rs.scenario_id + and connected[0][1] == rs.resource_attr_id): + raise HydraError( + "Refusing in-place update of dataset %s for resource attribute " + "%s in scenario %s: bulk_ctx.dataset_rs_map shows it is not " + "exclusively owned by this resource scenario (connected=%s)." % + (rs.dataset.id, rs.resource_attr_id, rs.scenario_id, connected)) + dataset = rs.dataset dataset.type = data_type dataset.value = val @@ -1323,13 +1388,23 @@ def assign_value(rs, data_type, val, # dataset_hash_cache is pre-seeded with DB-existing datasets that share a # candidate hash, and is updated during the loop to catch batch-internal # collisions (two update-path datasets converging on the same hash). - if dataset_hash_cache is not None: - existing = dataset_hash_cache.get(new_hash) - if existing is not None and existing.id != dataset.id and existing.check_read_permission(user_id, do_raise=False): - db.DBSession.delete(dataset) - dataset = existing - else: - dataset_hash_cache[new_hash] = dataset + existing = bulk_ctx.dataset_hash_cache.get(new_hash) + if existing is not None and existing.id != dataset.id and existing.check_read_permission(user_id, do_raise=False): + db.DBSession.delete(dataset) + dataset = existing + bulk_ctx.collisions_avoided += 1 + elif existing is not None and existing.id != dataset.id: + #Found a hash match we can't reuse (no read permission on it), and + #can't keep this hash either -- tDataset.hash has a UNIQUE + #constraint, so leaving it as-is would raise IntegrityError on + #flush. set_unique_hash() salts the hash computation only, it + #does not touch this dataset's real (persisted) metadata. + new_hash = dataset.set_unique_hash(metadata) + bulk_ctx.dataset_hash_cache[new_hash] = dataset + else: + bulk_ctx.dataset_hash_cache[new_hash] = dataset + + bulk_ctx.updated_in_place += 1 else: dataset = data.update_dataset(rs.dataset.id, name, data_type, val, unit_id, metadata, flush=False, **dict(user_id=user_id)) log.debug("Updated dataset '%s'", name) @@ -1338,8 +1413,8 @@ def assign_value(rs, data_type, val, log.debug("Set RS dataset id to %s"%dataset.id) else: log.debug("Creating new dataset %s in scenario %s", name, rs.scenario_id) - if new_dataset_cache is not None and data_hash in new_dataset_cache: - dataset = new_dataset_cache[data_hash] + if bulk_ctx is not None and data_hash in bulk_ctx.new_dataset_cache: + dataset = bulk_ctx.new_dataset_cache[data_hash] else: dataset = data.add_dataset( data_type, @@ -1349,8 +1424,10 @@ def assign_value(rs, data_type, val, name=name, **dict(user_id=user_id) ) - if new_dataset_cache is not None and data_hash is not None: - new_dataset_cache[data_hash] = dataset + if bulk_ctx is not None: + bulk_ctx.created += 1 + if data_hash is not None: + bulk_ctx.new_dataset_cache[data_hash] = dataset rs.dataset = dataset rs.source = source diff --git a/tests/test_scenario.py b/tests/test_scenario.py index 9ba6574c..016e8e5c 100644 --- a/tests/test_scenario.py +++ b/tests/test_scenario.py @@ -529,7 +529,187 @@ def test_bulk_update_resourcedata(self, client, network_with_data): if ra_id == descriptor['resource_attr_id']: assert rs.dataset.value == descriptor.dataset.value + def test_bulk_update_resourcedata_skips_unchanged_values(self, client, network_with_data): + """ + Regression test for the pre-pass "unchanged" short-circuit in + bulk_update_resourcedata: resubmitting the exact same value a resource + scenario already has must not create a new dataset or touch the + existing one. + """ + client.user_id = pytest.root_user_id + network1 = network_with_data + scenario = network1.scenarios[0] + + node = network1.nodes[5] + ra = client.testutils.get_by_name('node_attr_a', node.attributes) + + baseline_value = 111.111 + client.bulk_update_resourcedata( + [scenario.id], [client.testutils.create_scalar(ra, val=baseline_value)]) + + baseline_scenario = client.get_scenario(scenario.id) + baseline_rs = next(rs for rs in baseline_scenario.resourcescenarios + if rs.resource_attr_id == ra['id']) + baseline_dataset_id = baseline_rs.dataset.id + assert float(baseline_rs.dataset.value) == baseline_value + + #Resubmit the exact same value -- the pre-pass hash check should + #short-circuit this RA: no new dataset, existing one left untouched. + client.bulk_update_resourcedata( + [scenario.id], [client.testutils.create_scalar(ra, val=baseline_value)]) + + after_scenario = client.get_scenario(scenario.id) + after_rs = next(rs for rs in after_scenario.resourcescenarios + if rs.resource_attr_id == ra['id']) + assert after_rs.dataset.id == baseline_dataset_id, ( + "Resubmitting an unchanged value should not create a new dataset -- " + "the resource scenario should still point at the original one." + ) + assert float(after_rs.dataset.value) == baseline_value + + def test_bulk_update_resourcedata_dedupes_in_batch_hash_collision(self, client, network_with_data): + """ + Regression test for the batch-scoped dataset_hash_cache in assign_value's + bulk fast lane: two resource scenarios updated to the same new value in + the *same* bulk_update_resourcedata call must end up sharing a single + dataset, not each creating their own duplicate with an identical hash. + """ + client.user_id = pytest.root_user_id + network1 = network_with_data + scenario = network1.scenarios[0] + + node_a, node_b = network1.nodes[0], network1.nodes[1] + ra_a = client.testutils.get_by_name('node_attr_a', node_a.attributes) + ra_b = client.testutils.get_by_name('node_attr_a', node_b.attributes) + + #Give each RA its own distinct, exclusively-owned dataset first, so the + #collision below is guaranteed to go through the update-in-place fast + #lane (dataset_hash_cache), not the separately-tested new-dataset path. + client.bulk_update_resourcedata( + [scenario.id], + [client.testutils.create_scalar(ra_a, val=1.0), + client.testutils.create_scalar(ra_b, val=2.0)]) + + shared_value = 987654.321 + client.bulk_update_resourcedata( + [scenario.id], + [client.testutils.create_scalar(ra_a, val=shared_value), + client.testutils.create_scalar(ra_b, val=shared_value)]) + updated_scenario = client.get_scenario(scenario.id) + ds_a = next(rs.dataset for rs in updated_scenario.resourcescenarios + if rs.resource_attr_id == ra_a['id']) + ds_b = next(rs.dataset for rs in updated_scenario.resourcescenarios + if rs.resource_attr_id == ra_b['id']) + + assert float(ds_a.value) == shared_value + assert float(ds_b.value) == shared_value + assert ds_a.id == ds_b.id, ( + "Two resource scenarios updated to the same value in the same batch " + "should be deduplicated onto a single dataset via the in-batch " + "hash-collision cache." + ) + + def test_bulk_update_resourcedata_hash_collision_respects_dataset_permission( + self, client, network_with_data): + """ + Security regression test: the batch hash-collision cache must not + attach a resource scenario to an existing dataset the calling user + doesn't have read permission on, even though its hash matches. + """ + client.user_id = pytest.root_user_id + network1 = network_with_data + scenario = network1.scenarios[0] + + owner_node = network1.nodes[2] + ra_owner = client.testutils.get_by_name('node_attr_a', owner_node.attributes) + + secret_value = 424242.42 + client.bulk_update_resourcedata( + [scenario.id], [client.testutils.create_scalar(ra_owner, val=secret_value)]) + + owner_scenario = client.get_scenario(scenario.id) + hidden_dataset_id = next( + rs.dataset.id for rs in owner_scenario.resourcescenarios + if rs.resource_attr_id == ra_owner['id']) + + #Hide the dataset with no exceptions -- only its owner (root) can read it. + client.hide_dataset(hidden_dataset_id, [], 'N', 'N', 'N') + + #Share the network with UserC (non-admin) so they can legitimately edit + #it, but they have no read access to root's now-hidden dataset. + client.share_network(network1.id, ["UserC"], 'N', 'N') + + try: + client.user_id = pytest.user_c.id + other_node = network1.nodes[3] + ra_other = client.testutils.get_by_name('node_attr_a', other_node.attributes) + client.bulk_update_resourcedata( + [scenario.id], [client.testutils.create_scalar(ra_other, val=secret_value)]) + + client.user_id = pytest.root_user_id + after_scenario = client.get_scenario(scenario.id) + ds_other_id = next( + rs.dataset.id for rs in after_scenario.resourcescenarios + if rs.resource_attr_id == ra_other['id']) + + assert ds_other_id != hidden_dataset_id, ( + "UserC has no read permission on root's hidden dataset -- the " + "batch collision cache must not silently attach it to UserC's " + "resource scenario just because the hash matches." + ) + finally: + client.user_id = pytest.root_user_id + + def test_bulk_update_resourcedata_hash_collision_reuses_when_permitted( + self, client, network_with_data): + """ + Mirror of test_..._respects_dataset_permission: when the calling user + DOES have read permission on the matching-hash dataset, the batch + collision cache should reuse it rather than creating a duplicate. + """ + client.user_id = pytest.root_user_id + network1 = network_with_data + scenario = network1.scenarios[0] + + owner_node = network1.nodes[2] + ra_owner = client.testutils.get_by_name('node_attr_a', owner_node.attributes) + + shared_secret_value = 135791.113 + client.bulk_update_resourcedata( + [scenario.id], [client.testutils.create_scalar(ra_owner, val=shared_secret_value)]) + + owner_scenario = client.get_scenario(scenario.id) + target_dataset_id = next( + rs.dataset.id for rs in owner_scenario.resourcescenarios + if rs.resource_attr_id == ra_owner['id']) + + #Hide the dataset, but explicitly grant UserC read access to it. + client.hide_dataset(target_dataset_id, ["UserC"], 'Y', 'N', 'N') + + client.share_network(network1.id, ["UserC"], 'N', 'N') + + try: + client.user_id = pytest.user_c.id + other_node = network1.nodes[3] + ra_other = client.testutils.get_by_name('node_attr_a', other_node.attributes) + client.bulk_update_resourcedata( + [scenario.id], + [client.testutils.create_scalar(ra_other, val=shared_secret_value)]) + + client.user_id = pytest.root_user_id + after_scenario = client.get_scenario(scenario.id) + ds_other_id = next( + rs.dataset.id for rs in after_scenario.resourcescenarios + if rs.resource_attr_id == ra_other['id']) + + assert ds_other_id == target_dataset_id, ( + "UserC has read permission on the matching-hash dataset -- the " + "batch collision cache should reuse it instead of creating a " + "duplicate." + ) + finally: + client.user_id = pytest.root_user_id def test_bulk_add_data(self, client, dateformat): From 843eac9a30e6bd7733b923a26e79c42ef4b4b93a Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Thu, 9 Jul 2026 14:46:33 +0000 Subject: [PATCH 12/15] Fix alembic head divergence from merged node_alt_coords migration b7f3e1a92c44 was branched off a1b2c3d4e5f6 before edf7bffb7b33 (this branch's merge of the divergent a1b2c3d4e5f6/580425ade2e4/877adf863b33 heads) existed, so merging both in left two heads again. Repoint it onto edf7bffb7b33 to restore a single alembic head. --- .../db/alembic/versions/b7f3e1a92c44_node_alt_coords.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/hydra_base/db/alembic/versions/b7f3e1a92c44_node_alt_coords.py b/hydra_base/db/alembic/versions/b7f3e1a92c44_node_alt_coords.py index 15974505..0be0674b 100644 --- a/hydra_base/db/alembic/versions/b7f3e1a92c44_node_alt_coords.py +++ b/hydra_base/db/alembic/versions/b7f3e1a92c44_node_alt_coords.py @@ -1,7 +1,7 @@ """node_alt_coords Revision ID: b7f3e1a92c44 -Revises: a1b2c3d4e5f6 +Revises: edf7bffb7b33 Create Date: 2026-07-09 00:00:00.000000 """ @@ -13,7 +13,7 @@ # revision identifiers, used by Alembic. revision = 'b7f3e1a92c44' -down_revision = 'a1b2c3d4e5f6' +down_revision = 'edf7bffb7b33' branch_labels = None depends_on = None From feecc7980cfe3ad8d124fbf9c1522ae4a884a489 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Thu, 9 Jul 2026 15:52:48 +0000 Subject: [PATCH 13/15] fix: catch memcached errors in build_user_cache cache check The direct cache.get() call here bypassed get_cache()'s error handling, so a memcached timeout (e.g. from a pod restart) propagated as an unhandled 500 instead of degrading gracefully like the other cache accessors on this class. --- hydra_base/db/model/project.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/hydra_base/db/model/project.py b/hydra_base/db/model/project.py index 8c7ee45c..f1467328 100644 --- a/hydra_base/db/model/project.py +++ b/hydra_base/db/model/project.py @@ -274,8 +274,11 @@ def build_user_cache(cls, uid): Build the cache of projects a user has access to either by direct Ownership or by indirect access required for navigating to a project to which they own """ - if cache.get(_user_project_cache_key(uid)) is not None: - return + try: + if cache.get(_user_project_cache_key(uid)) is not None: + return + except Exception as e: + log.warning(f"Error checking project cache for user {uid}: {e}") user_cache = defaultdict(list) projects_qry = get_session().query(Project) From 6874f6174fa08b7845d100e820dc99c794025cf2 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Mon, 13 Jul 2026 16:18:35 +0000 Subject: [PATCH 14/15] Add a function and associated test to update the appdata on a network --- hydra_base/lib/network.py | 42 ++++++++++++++++++++++++++++++++-- tests/test_network.py | 48 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 2 deletions(-) diff --git a/hydra_base/lib/network.py b/hydra_base/lib/network.py index e1aa3064..7566a358 100644 --- a/hydra_base/lib/network.py +++ b/hydra_base/lib/network.py @@ -1665,6 +1665,35 @@ def update_resource_layout(resource_type, resource_id, key, value, **kwargs): return layout +def update_network_appdata(network_id, key, value, **kwargs): + """ + Update a single key in a network's appdata without touching any + other network fields (name, description, projection, layout, etc). + This assumes that appdata is a JSON compatible dictionary. + """ + user_id = kwargs.get('user_id') + + log.info("Updating network %s's appdata with {%s:%s}", network_id, key, value) + + try: + net_i = db.DBSession.query(Network).filter(Network.id == network_id).one() + except NoResultFound: + raise ResourceNotFoundError("Network with id %s not found"%(network_id)) + + net_i.check_write_permission(user_id) + + if net_i.appdata is None: + appdata = dict() + else: + appdata = json.loads(net_i.appdata) + + appdata[key] = value + net_i.appdata = json.dumps(appdata) + + db.DBSession.flush() + + return appdata + def get_resource(resource_type, resource_id, **kwargs): user_id = kwargs.get('user_id') @@ -1715,7 +1744,9 @@ def get_network_extents(network_id,**kwargs): min_alt_x=None, max_alt_x=None, min_alt_y=None, - max_alt_y=None + max_alt_y=None, + has_geographic=False, + has_schematic=False ) # Compute min/max extent of the network. @@ -1761,7 +1792,14 @@ def get_network_extents(network_id,**kwargs): min_alt_x=min_alt_x, max_alt_x=max_alt_x, min_alt_y=min_alt_y, - max_alt_y=max_alt_y + max_alt_y=max_alt_y, + # `min`/`max` above default to a fake (0, 1) range when no node in + # the network has that coordinate system populated, so callers that + # need to know whether the coordinate system is actually usable + # (e.g. to decide whether to offer a map/schematic view) must check + # these flags rather than the min/max values themselves. + has_geographic=len(x) > 0 and len(y) > 0, + has_schematic=len(alt_x) > 0 and len(alt_y) > 0 )) return ne diff --git a/tests/test_network.py b/tests/test_network.py index ee8336e4..dc4dd0cd 100644 --- a/tests/test_network.py +++ b/tests/test_network.py @@ -302,6 +302,54 @@ def test_get_extents(self, client, network_with_data): assert extents.min_y == 9 assert extents.max_y == 99 + def test_update_network_appdata(self, client, network_with_data): + """ + Test that a single key can be added to a network's appdata without + disturbing any keys already present. + """ + net = network_with_data + + appdata = client.update_network_appdata(net.id, 'foo', 'bar') + assert appdata['foo'] == 'bar' + + stored_net = client.get_network(net.id) + assert json.loads(stored_net.appdata) == {'foo': 'bar'} + + #Updating a different key should leave the existing one untouched + appdata = client.update_network_appdata(net.id, 'baz', {'nested': 1}) + assert appdata == {'foo': 'bar', 'baz': {'nested': 1}} + + stored_net = client.get_network(net.id) + assert json.loads(stored_net.appdata) == {'foo': 'bar', 'baz': {'nested': 1}} + + #Updating an existing key should overwrite its value + appdata = client.update_network_appdata(net.id, 'foo', 'updated') + assert appdata == {'foo': 'updated', 'baz': {'nested': 1}} + + def test_update_network_appdata_unknown_network(self, client): + """ + Updating the appdata of a network which does not exist should raise + an error. + """ + with pytest.raises(hb.exceptions.HydraError): + client.update_network_appdata(999999, 'foo', 'bar') + + def test_update_network_appdata_no_permission(self, client, networkmaker): + """ + A user without write access to a network should not be able to + update its appdata. + """ + net = networkmaker.create() + + #UserD is not shared onto the default test project (unlike UserA/B/C), + #so it has no view/edit access to this network. + client.login('UserD', 'password') + try: + with pytest.raises(hb.exceptions.HydraError): + client.update_network_appdata(net.id, 'foo', 'bar') + finally: + client.login('root', '') + def test_update_network(self, client, network_with_data): net = hb.JSONObject(client.get_network(network_with_data.id)) From a4e233a7fe090dc37d10bcda4cbc7d2585a1ec88 Mon Sep 17 00:00:00 2001 From: Stephen Knox Date: Mon, 13 Jul 2026 16:22:23 +0000 Subject: [PATCH 15/15] Add an update_appdata function for a project and the associated test --- hydra_base/lib/project.py | 26 +++++++++++++++++++ tests/project/test_project.py | 47 +++++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+) diff --git a/hydra_base/lib/project.py b/hydra_base/lib/project.py index d5f6459b..d49f17b5 100644 --- a/hydra_base/lib/project.py +++ b/hydra_base/lib/project.py @@ -172,6 +172,32 @@ def update_project(project, **kwargs): return proj_i +def update_project_appdata(project_id, key, value, **kwargs): + """ + Update a single key in a project's appdata without touching any + other project fields (name, description, parent_id, etc). + This assumes that appdata is a JSON compatible dictionary. + """ + user_id = kwargs.get('user_id') + + log.info("Updating project %s's appdata with {%s:%s}", project_id, key, value) + + proj_i = _get_project(project_id, user_id, check_write=True) + + if proj_i.appdata is None: + appdata = dict() + else: + appdata = proj_i.appdata.copy() + + appdata[key] = value + proj_i.appdata = appdata + + Project.clear_cache(user_id) + + db.DBSession.flush() + + return appdata + @required_perms('edit_project') def move_project(project_id, target_project_id, **kwargs): """ diff --git a/tests/project/test_project.py b/tests/project/test_project.py index 5dd3b45d..b5bfc913 100644 --- a/tests/project/test_project.py +++ b/tests/project/test_project.py @@ -126,6 +126,53 @@ def test_update(self, client, network_with_data): rs_to_check.dataset.value == 'just project desscriptor', \ "There is an inconsistency with the attributes." + def test_update_project_appdata(self, client, projectmaker): + """ + Test that a single key can be added to a project's appdata without + disturbing any keys already present. + """ + proj = projectmaker.create() + + appdata = client.update_project_appdata(proj.id, 'foo', 'bar') + assert appdata['foo'] == 'bar' + + stored_proj = client.get_project(proj.id) + assert stored_proj.appdata == {'foo': 'bar'} + + #Updating a different key should leave the existing one untouched + appdata = client.update_project_appdata(proj.id, 'baz', {'nested': 1}) + assert appdata == {'foo': 'bar', 'baz': {'nested': 1}} + + stored_proj = client.get_project(proj.id) + assert stored_proj.appdata == {'foo': 'bar', 'baz': {'nested': 1}} + + #Updating an existing key should overwrite its value + appdata = client.update_project_appdata(proj.id, 'foo', 'updated') + assert appdata == {'foo': 'updated', 'baz': {'nested': 1}} + + def test_update_project_appdata_unknown_project(self, client): + """ + Updating the appdata of a project which does not exist should raise + an error. + """ + with pytest.raises(hb.exceptions.HydraError): + client.update_project_appdata(999999, 'foo', 'bar') + + def test_update_project_appdata_no_permission(self, client, projectmaker): + """ + A user without write access to a project should not be able to + update its appdata. + """ + proj_user = client.user_id + proj = projectmaker.create(share=False) + + client.user_id = pytest.user_c.id + try: + with pytest.raises(hb.exceptions.HydraError): + client.update_project_appdata(proj.id, 'foo', 'bar') + finally: + client.user_id = proj_user + def test_load(self, client): project = JSONObject({}) project.name = 'Test Project %s'%(datetime.datetime.now())