"""Provides workflows for managing queries within iMednet studies."""
from typing import TYPE_CHECKING, Any
from imednet.spi.models import Query
if TYPE_CHECKING:
from imednet.spi.facade import ImednetFacade
[docs]class QueryManagementWorkflow:
"""Provides methods for common query management tasks.
Args:
sdk: An instance of the ImednetSDK.
"""
[docs] def __init__(self, sdk: "ImednetFacade"):
"""Initialize the query management workflow.
Args:
sdk: An instance of the iMednet SDK facade.
"""
self._sdk = sdk
@staticmethod
def _is_query_open(query: Query) -> bool | None:
"""Determines if a query is open based on its comments.
Returns None if the state cannot be determined (no comments).
"""
if not getattr(
query,
'query_comments',
getattr(query, 'model_extra', {}).get('query_comments', [])
if getattr(query, 'model_extra', None) is not None
else [],
):
return None
# Find the comment with the highest sequence number
latest_comment = max(
getattr(
query,
'query_comments',
getattr(query, 'model_extra', {}).get('query_comments', [])
if getattr(query, 'model_extra', None) is not None
else [],
),
key=lambda c: c.sequence,
)
# Check if the latest comment indicates the query is open (closed=False)
return not latest_comment.closed
[docs] def get_open_queries(
self,
study_key: str,
additional_filter: dict[str, Any | tuple[str, Any] | list[Any]] | None = None,
**kwargs: Any,
) -> list[Query]:
"""Retrieves all open queries for a given study, potentially filtered further.
An 'open' query is defined as one where the query comment with the highest
sequence number has its 'closed' field set to False.
Note:
This method fetches queries based on `additional_filter` and then
filters for the 'open' state client-side.
Args:
study_key: The key identifying the study.
additional_filter: An optional dictionary of conditions to apply via the API.
**kwargs: Additional keyword arguments passed directly to `sdk.queries.list`.
Returns:
A list of open Query objects matching the criteria.
"""
filters = dict(additional_filter) if additional_filter else {}
# Fetch potentially relevant queries
all_matching_queries = self._sdk.get_queries(study_key, **filters, **kwargs)
open_queries: list[Query] = []
for query in all_matching_queries:
if self._is_query_open(query) is True:
open_queries.append(query)
return open_queries
[docs] def get_queries_for_subject(
self,
study_key: str,
subject_key: str,
additional_filter: dict[str, Any | tuple[str, Any] | list[Any]] | None = None,
**kwargs: Any,
) -> list[Query]:
"""Retrieves all queries for a specific subject within a study.
Args:
study_key: The key identifying the study.
subject_key: The key identifying the subject.
additional_filter: An optional dictionary of conditions to combine
with the subject filter.
**kwargs: Additional keyword arguments passed directly to `sdk.queries.list`.
Returns:
A list of Query objects for the specified subject.
"""
# Build filter dictionary
final_filter_dict: dict[str, Any] = {"subject_key": subject_key}
if additional_filter:
final_filter_dict.update(additional_filter)
return list(self._sdk.get_queries(study_key, **final_filter_dict, **kwargs))
[docs] def get_queries_by_site(
self,
study_key: str,
site_key: str,
additional_filter: dict[str, Any | tuple[str, Any] | list[Any]] | None = None,
**kwargs: Any,
) -> list[Query]:
"""Retrieves all queries for a specific site within a study.
Args:
study_key: The key identifying the study.
site_key: The name of the site.
additional_filter: Extra conditions to combine with the subject filter.
**kwargs: Additional keyword arguments passed directly to `sdk.queries.list`.
Returns:
A list of Query objects for the specified site.
"""
subjects = self._sdk.get_subjects(study_key, site_name=site_key)
subject_keys = [s.subject_key for s in subjects]
if not subject_keys:
return []
final_filter_dict: dict[str, Any] = {"subject_key": subject_keys}
if additional_filter:
final_filter_dict.update(additional_filter)
return list(self._sdk.get_queries(study_key, **final_filter_dict, **kwargs))
[docs] def get_query_state_counts(self, study_key: str, **kwargs: Any) -> dict[str, int]:
"""Counts queries grouped by their current state (open/closed/unknown).
The state is determined by the 'closed' field of the query comment
with the highest sequence number. Queries without any comments are
counted as 'unknown'.
Note:
This method fetches all queries matching the base criteria (if any
are passed via kwargs) and then performs the aggregation client-side.
Args:
study_key: The key identifying the study.
**kwargs: Additional keyword arguments passed directly to `sdk.queries.list`
(e.g., for initial filtering before counting).
Returns:
A dictionary with keys 'open', 'closed', 'unknown' and their respective counts.
"""
all_queries = self._sdk.get_queries(study_key, **kwargs)
# Initialize counts for all possible states
state_counts: dict[str, int] = {"open": 0, "closed": 0, "unknown": 0}
for query in all_queries:
is_open = self._is_query_open(query)
if is_open is None:
state_counts["unknown"] += 1
elif is_open:
state_counts["open"] += 1
else:
state_counts["closed"] += 1
return state_counts
# Integration:
# - Accessed via the main SDK instance
# (e.g., `sdk.workflows.query_management.get_open_queries(...)`).
# - Provides convenient ways to access query information without manually constructing
# complex filters.