refactor(metastore): rework metastore (sql store) methods

[AlessandroPomponio]

• Support field-level querying (selects)
• [ ]

[AlessandroPomponio]

SQLResourceStore Methods Diagram

I got Bob to create a mermaid diagram of the methods in SQLResourceStore

classDiagram
    class SQLResourceStore {
        +__init__(project_context, ensureExists)
        +__getstate__() dict
        +__setstate__(state)
        +engine() Engine
        +get_resource_and_producers(identifier, kind, chain, raise_error) list
        +getResourceRaw(identifier) dict
        +getResource(identifier, kind, raise_error) ADOResource
        +getResources(identifiers, ignore_validation_errors) dict
        +getResourcesOfKind(kind, version, field_selectors, ignore_errors) dict
        +getResourceIdentifiersOfKind(kind, version, field_selectors, details) DataFrame
        +get_latest_resource_identifiers_of_kinds(kinds) dict
        +resourceTable() DataFrame
        +getRelatedSubjectResourceIdentifiers(identifier, kind, version) tuple
        +getRelatedObjectResourceIdentifiers(identifier, kind, version) tuple
        +getRelatedResourceIdentifiers(identifier, kind, version) tuple
        +getRelatedResources(identifier, kind) dict
        +containsResourceWithIdentifier(identifier, kind) bool
        +addResource(resource)
        +addRelationship(subjectIdentifier, objectIdentifier)
        +addRelationshipForResources(subjectResource, objectResource)
        +addResourceWithRelationships(resource, relatedIdentifiers)
        +updateResource(resource)
        +deleteResource(identifier)
        +deleteObjectRelationships(identifier)
        +delete_sample_store(identifier, force_deletion)
        +delete_operation(identifier, ignore_running_operations)
        +delete_discovery_space(identifier)
        +delete_data_container(identifier)
        +delete_actuator_configuration(identifier)
    }

Loading

Alternative View: Method Flow Diagram

flowchart TB
    subgraph Lifecycle["Lifecycle Methods"]
        init["__init__()"]
        getstate["__getstate__()"]
        setstate["__setstate__()"]
        engine["engine property"]
    end

    subgraph SingleRetrieval["Single Resource Retrieval"]
        getResource["getResource()"]
        getResourceRaw["getResourceRaw()"]
        getChain["get_resource_and_producers()"]
    end

    subgraph MultiRetrieval["Multiple Resource Retrieval"]
        getResources["getResources()"]
        getResourcesOfKind["getResourcesOfKind()"]
        getIdentifiers["getResourceIdentifiersOfKind()"]
        getLatest["get_latest_resource_identifiers_of_kinds()"]
        resourceTable["resourceTable()"]
    end

    subgraph Relationships["Relationship Queries"]
        getSubject["getRelatedSubjectResourceIdentifiers()"]
        getObject["getRelatedObjectResourceIdentifiers()"]
        getRelated["getRelatedResourceIdentifiers()"]
        getRelatedRes["getRelatedResources()"]
    end

    subgraph Existence["Existence Checks"]
        contains["containsResourceWithIdentifier()"]
    end

    subgraph Create["Resource Creation"]
        addResource["addResource()"]
        addRelationship["addRelationship()"]
        addRelForRes["addRelationshipForResources()"]
        addWithRel["addResourceWithRelationships()"]
    end

    subgraph Update["Resource Updates"]
        updateResource["updateResource()"]
    end

    subgraph Delete["Resource Deletion"]
        deleteResource["deleteResource()"]
        deleteObjRel["deleteObjectRelationships()"]
        deleteSample["delete_sample_store()"]
        deleteOp["delete_operation()"]
        deleteSpace["delete_discovery_space()"]
        deleteContainer["delete_data_container()"]
        deleteActuator["delete_actuator_configuration()"]
    end

    style Lifecycle fill:#e1f5ff
    style SingleRetrieval fill:#fff4e1
    style MultiRetrieval fill:#ffe1f5
    style Relationships fill:#e1ffe1
    style Existence fill:#f5e1ff
    style Create fill:#ffe1e1
    style Update fill:#e1e1ff
    style Delete fill:#ffffe1

Loading

Method Categories

1. Lifecycle & Configuration

• __init__(project_context, ensureExists) - Initialize the SQL resource store
• __getstate__() - Prepare for pickling (remove non-picklable engine)
• __setstate__(state) - Restore from pickle (recreate engine)
• engine - Property to access SQLAlchemy engine

2. Single Resource Retrieval

• getResource(identifier, kind, raise_error_if_no_resource) - Get a single
  resource by identifier and kind
• getResourceRaw(identifier) - Get raw JSON data without validation
• get_resource_and_producers(identifier, kind, chain, raise_error_if_no_resource) -
  Fetch resource and producer chain in single JOIN

3. Multiple Resource Retrieval

• getResources(identifiers, ignore_validation_errors) - Get multiple resources
  by identifiers
• getResourcesOfKind(kind, version, field_selectors, ignore_validation_errors) -
  Get all resources of a specific kind
• getResourceIdentifiersOfKind(kind, version, field_selectors, details) - Get
  identifiers with optional filtering and details
• get_latest_resource_identifiers_of_kinds(kinds) - Get latest resource
  identifiers for multiple kinds
• resourceTable() - Get all resources as DataFrame

4. Relationship Queries

• getRelatedSubjectResourceIdentifiers(identifier, kind, version) - Get
  resources where this is the subject
• getRelatedObjectResourceIdentifiers(identifier, kind, version) - Get
  resources where this is the object
• getRelatedResourceIdentifiers(identifier, kind, version) - Get all related
  resources (both directions)
• getRelatedResources(identifier, kind) - Get related resource objects

5. Existence Checks

• containsResourceWithIdentifier(identifier, kind) - Check if resource exists

6. Resource Creation

• addResource(resource) - Add a new resource
• addRelationship(subjectIdentifier, objectIdentifier) - Create relationship
  between resources
• addRelationshipForResources(subjectResource, objectResource) - Create
  relationship using resource objects
• addResourceWithRelationships(resource, relatedIdentifiers) - Add resource
  and create relationships

7. Resource Updates

• updateResource(resource) - Update an existing resource

8. Resource Deletion

• deleteResource(identifier) - Delete a resource
• deleteObjectRelationships(identifier) - Delete relationships where resource
  is object
• delete_sample_store(identifier, force_deletion) - Delete sample store and
  associated tables
• delete_operation(identifier, ignore_running_operations) - Delete operation
  with cleanup
• delete_discovery_space(identifier) - Delete discovery space
• delete_data_container(identifier) - Delete data container
• delete_actuator_configuration(identifier) - Delete actuator configuration

Key Design Patterns

1. Pickle Support: Custom __getstate__ and __setstate__ for Ray
   compatibility
2. Flexible Retrieval: Multiple methods for different retrieval patterns
   (single, multiple, by kind, with relationships)
3. Relationship Management: Bidirectional relationship queries
   (subject/object)
4. Type-Specific Deletion: Specialized delete methods for each resource type
   with proper cleanup
5. Error Handling: Optional error raising for missing resources
6. Validation Control: Optional validation error handling for bulk
   operations

[AlessandroPomponio]

I had a very long design discussion with Copilot and we converged to having these Pydantic models (slight changes might need to still be done):

from __future__ import annotations

from enum import Enum
from typing import Annotated, Any, Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator


# ---------------------------------------------------------------------------
# Common type aliases
# ---------------------------------------------------------------------------

Identifier = str
JsonPath = str
ResourceKind = str
ResourceVersion = str


# ---------------------------------------------------------------------------
# Reusable predicate model
# ---------------------------------------------------------------------------

class ValueCriterion(BaseModel):
    """
    Reusable predicate model for filtering values.

    Suggested semantics:
    - multiple populated operators are combined with AND
    - some operator combinations are rejected as ambiguous or contradictory
    """
    model_config = ConfigDict(extra="forbid")

    # Equality / inequality
    eq: Any | None = None
    ne: Any | None = None

    # Membership
    one_of: list[Any] | None = None
    not_in: list[Any] | None = None

    # Comparison
    gt: Any | None = None
    gte: Any | None = None
    lt: Any | None = None
    lte: Any | None = None

    # Text matching
    like: str | None = None
    ilike: str | None = None
    starts_with: str | None = None
    ends_with: str | None = None
    contains: str | None = None

    # Presence / nullability
    exists: bool | None = None
    is_null: bool | None = None

    @model_validator(mode="after")
    def validate_criterion(self) -> "ValueCriterion":
        populated = [
            name for name, value in self.__dict__.items()
            if value is not None
        ]
        if not populated:
            raise ValueError("ValueCriterion must define at least one operator")

        if self.gt is not None and self.gte is not None:
            raise ValueError("gt and gte are mutually exclusive")

        if self.lt is not None and self.lte is not None:
            raise ValueError("lt and lte are mutually exclusive")

        text_ops = [
            name for name in ["like", "ilike", "starts_with", "ends_with", "contains"]
            if getattr(self, name) is not None
        ]
        if len(text_ops) > 1:
            raise ValueError(
                "Only one text-matching operator can be used at a time"
            )

        if self.eq is not None and self.one_of is not None:
            raise ValueError("eq and one_of should not be combined")

        if self.exists is False:
            other_ops = [name for name in populated if name != "exists"]
            if other_ops:
                raise ValueError(
                    "exists=False cannot be combined with other operators"
                )

        if self.is_null is True:
            other_ops = [name for name in populated if name not in {"is_null", "exists"}]
            if other_ops:
                raise ValueError(
                    "is_null=True cannot be combined with value-comparison operators"
                )

        return self


# ---------------------------------------------------------------------------
# Arbitrary field filtering
# ---------------------------------------------------------------------------

class FieldCriterion(BaseModel):
    """
    Apply a ValueCriterion to an arbitrary field or JSON path.

    Examples:
    - "kind"
    - "version"
    - "$.config.sampleStoreIdentifier"
    - "$.status.phase"
    """
    model_config = ConfigDict(extra="forbid")

    field: str
    criterion: ValueCriterion


# ---------------------------------------------------------------------------
# Root filtering
# ---------------------------------------------------------------------------

class StoreFilter(BaseModel):
    """
    Semantics:
    - within one StoreFilter: AND
    - across list[StoreFilter]: OR
    """
    model_config = ConfigDict(extra="forbid")

    identifier: ValueCriterion | None = None
    kind: ValueCriterion | None = None
    version: ValueCriterion | None = None
    latest: bool | None = None

    labels: dict[str, ValueCriterion] = Field(default_factory=dict)
    metadata: dict[str, ValueCriterion] = Field(default_factory=dict)

    # arbitrary field / JSONPath filters
    fields: list[FieldCriterion] = Field(default_factory=list)


# ---------------------------------------------------------------------------
# Projection
# ---------------------------------------------------------------------------

class ProjectionField(BaseModel):
    """
    Select a field from the resource JSON/document and optionally rename it.
    """
    model_config = ConfigDict(extra="forbid")

    path: JsonPath
    alias: str | None = None


class ProjectionSpec(BaseModel):
    """
    Projection definition.

    If projection is provided, callers should generally expect dict-like results
    rather than full validated resource models.
    """
    model_config = ConfigDict(extra="forbid")

    fields: list[ProjectionField] = Field(default_factory=list)


# ---------------------------------------------------------------------------
# Query options
# ---------------------------------------------------------------------------

class SortDirection(str, Enum):
    ASC = "asc"
    DESC = "desc"


class OrderBy(BaseModel):
    """
    Sort by a logical field or JSONPath-like expression.
    """
    model_config = ConfigDict(extra="forbid")

    field: str
    direction: SortDirection = SortDirection.ASC


class GroupBy(BaseModel):
    """
    Group by a logical field or JSONPath-like expression.
    """
    model_config = ConfigDict(extra="forbid")

    field: str


class FindOptions(BaseModel):
    """
    Generic query execution/materialization options.
    """
    model_config = ConfigDict(extra="forbid")

    raw: bool = False
    ignore_validation_errors: bool = False

    limit: int | None = Field(default=None, ge=1)
    offset: int | None = Field(default=None, ge=0)

    order_by: list[OrderBy] = Field(default_factory=list)
    group_by: list[GroupBy] = Field(default_factory=list)


# ---------------------------------------------------------------------------
# Reference traversal
# ---------------------------------------------------------------------------

class ReferenceTraversalMode(str, Enum):
    FLAT = "flat"
    MAPPED = "mapped"


class ReferenceHop(BaseModel):
    """
    Resolve a referenced resource identifier from a field in the resource JSON.

    Example:
        path="$.config.sampleStoreIdentifier"
        kind="SampleStore"
        alias="sample_store"
    """
    model_config = ConfigDict(extra="forbid")

    path: JsonPath
    kind: ResourceKind | None = None
    version: ResourceVersion | None = None
    alias: str | None = None
    many: bool = False


class ReferenceTraversal(BaseModel):
    """
    Traverse by dereferencing identifiers found in JSON fields of the selected root resource(s).
    """
    model_config = ConfigDict(extra="forbid")

    type: Literal["reference"] = "reference"

    hops: list[ReferenceHop] = Field(default_factory=list)

    include_root: bool = True
    mode: ReferenceTraversalMode = ReferenceTraversalMode.MAPPED


# ---------------------------------------------------------------------------
# Relationship traversal
# ---------------------------------------------------------------------------

class RelationshipDirection(str, Enum):
    OUTBOUND = "outbound"   # subject -> object
    INBOUND = "inbound"     # object -> subject
    BOTH = "both"


class RelationshipReturnMode(str, Enum):
    NODES = "nodes"
    PATHS = "paths"


class RelationshipTraversal(BaseModel):
    """
    Traverse using the relationship table.
    """
    model_config = ConfigDict(extra="forbid")

    type: Literal["relationship"] = "relationship"

    direction: RelationshipDirection = RelationshipDirection.BOTH

    # depth=1 means one hop; depth=None means recursive / unbounded
    depth: int | None = Field(default=1, ge=1)

    include_root: bool = False
    deduplicate: bool = True

    # constraints on traversed / returned nodes
    kind: ValueCriterion | None = None
    version: ValueCriterion | None = None
    labels: dict[str, ValueCriterion] = Field(default_factory=dict)
    metadata: dict[str, ValueCriterion] = Field(default_factory=dict)
    fields: list[FieldCriterion] = Field(default_factory=list)

    return_mode: RelationshipReturnMode = RelationshipReturnMode.NODES


# ---------------------------------------------------------------------------
# Discriminated union for traversal
# ---------------------------------------------------------------------------

TraversalSpec = Annotated[
    ReferenceTraversal | RelationshipTraversal,
    Field(discriminator="type"),
]


# ---------------------------------------------------------------------------
# Main query model
# ---------------------------------------------------------------------------

class FindQuery(BaseModel):
    """
    Central read/query model.

    Suggested semantics:
    - identifiers restrict the root resource set
    - filters apply to the root resource set
    - projection shapes the output
    - options control execution/materialization
    - traversal optionally adds reference or relationship traversal semantics
    """
    model_config = ConfigDict(extra="forbid")

    identifiers: Identifier | set[Identifier] | None = None
    filters: list[StoreFilter] = Field(default_factory=list)
    projection: ProjectionSpec | None = None
    options: FindOptions = Field(default_factory=FindOptions)
    traversal: TraversalSpec | None = None

[AlessandroPomponio]

We could then have methods like

class ResourceStore:
    def find(self, query: FindQuery) -> list[Resource | dict]: ...
    def find_one(self, query: FindQuery) -> Resource | dict | None: ...
    def get(self, query: FindQuery) -> Resource | dict: ...
    def exists(self, query: FindQuery) -> bool: ...
    def find_ids(self, query: FindQuery) -> list...

    # Convenience APIs
    def get_resource(self, identifier: str, kind: str | None = None) -> Resource: ...
    def get_raw(self, identifier: str) -> dict: ...
    def find_related(self, identifier: str, direction: RelationshipDirection, ...) -> list[str]: ...

ofc with more convenience APIs. All the methods would map to the base find

[AlessandroPomponio]

To extend this to Update and Delete, we can extract identifiers and filters from FindQuery and make them reusable. We will then have to consider relationship semantics, cascading deletes, etc. but in general we should be fine. On Monday I'll double check