Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class CollectionValueGETResponse(ProtoModel):
class Data(ProtoModel):
values: bytes = Field(..., description="Feather-serialized bytes representing a pandas DataFrame.")
units: Dict[str, str] = Field(..., description="Units of value columns.")
meta: CollectionSubresourceGETResponseMeta = Field(
..., description=str(get_base_docs(CollectionSubresourceGETResponseMeta))
)
data: Optional[Data] = Field(..., description="Values and units.")
register_model("collection/[0-9]+/value", "GET", CollectionValueGETBody, CollectionValueGETResponse)
class CollectionListGETBody(ProtoModel):
class Data(ProtoModel):
pass
meta: EmptyMeta = Field(EmptyMeta(), description=common_docs[EmptyMeta])
data: Data = Field(..., description="Empty for now.")
class CollectionListGETResponse(ProtoModel):
meta: CollectionSubresourceGETResponseMeta = Field(
..., description=str(get_base_docs(CollectionSubresourceGETResponseMeta))
)
data: Optional[bytes] = Field(..., description="Feather-serialized bytes representing a pandas DataFrame.")
register_model("collection/[0-9]+/list", "GET", CollectionListGETBody, CollectionListGETResponse)
### Result
meta: QueryMeta = Field(QueryMeta(), description=common_docs[QueryMeta])
data: Data = Field(..., description="The keys with data to search the database on for Services.")
class ServiceQueueGETResponse(ProtoModel):
meta: ResponseGETMeta = Field(..., description=common_docs[ResponseGETMeta])
data: List[Dict[str, Optional[Any]]] = Field(
..., description="The return of Services found in the database mapping their Ids to the Service spec."
)
register_model("service_queue", "GET", ServiceQueueGETBody, ServiceQueueGETResponse)
class ServiceQueuePOSTBody(ProtoModel):
class Meta(ProtoModel):
tag: Optional[str] = Field(
None,
description="Tag to assign to the Tasks this Service will generate so that Queue Managers can pull only "
"Tasks based on this entry. If no Tag is specified, any Queue Manager can pull this Tasks "
"created by this Service.",
)
priority: Union[str, int, None] = Field(
None,
description="Priority given to this Tasks created by this Service. Higher priority will be pulled first.",
)
meta: Meta = Field(
...,
description="Metadata information for the Service for the Tag and Priority of Tasks this Service will create.",
)
data: List[Union[TorsionDriveInput, GridOptimizationInput]] = Field(
meta: ResponsePOSTMeta = Field(..., description=common_docs[ResponsePOSTMeta])
data: List[ObjectId] = Field(
...,
description="A list of Id's assigned to the Molecule objects passed in which serves as a unique identifier "
"in the database. If the Molecule was already in the database, then the Id returned is its "
"existing Id (entries are not duplicated).",
)
register_model("molecule", "POST", MoleculePOSTBody, MoleculePOSTResponse)
### Keywords
class KeywordGETBody(ProtoModel):
class Data(ProtoModel):
id: QueryObjectId = None
hash_index: QueryStr = None
meta: QueryMeta = Field(QueryMeta(), description=common_docs[QueryMeta])
data: Data = Field(
...,
description="The formal query for a Keyword fetch, contains ``id`` or ``hash_index`` for the object to fetch.",
)
class KeywordGETResponse(ProtoModel):
meta: ResponseGETMeta = Field(..., description=common_docs[ResponseGETMeta])
data: List[KeywordSet] = Field(
..., description="The :class:`KeywordSet` found from in the database based on the query."
)
import datetime
from enum import Enum
from typing import Any, Dict, List, Optional, Union
from pydantic import Field, validator
from qcelemental.models import ComputeError
from .common_models import ObjectId, ProtoModel
class DBRef(ProtoModel):
"""
Database locator reference object. Identifies an exact record in a database.
"""
ref: str = Field(..., description="The name of the table which the Database entry exists")
id: ObjectId = Field(..., description="The Database assigned Id of the entry in the ``ref`` table.")
class TaskStatusEnum(str, Enum):
"""
The state of a Task object. The states which are available are a finite set.
"""
running = "RUNNING"
waiting = "WAITING"
error = "ERROR"
class ProcedureGETResponse(ProtoModel):
meta: ResponseGETMeta = Field(..., description=common_docs[ResponseGETMeta])
data: List[Dict[str, Optional[Any]]] = Field(
..., description="The list of Procedure specs found based on the query."
)
register_model("procedure", "GET", ProcedureGETBody, ProcedureGETResponse)
### Task Queue
class TaskQueueGETBody(ProtoModel):
class Data(ProtoModel):
id: QueryObjectId = Field(
None,
description="The exact Id to fetch from the database. If this is set as a search condition, there is no "
"reason to set anything else as this will be unique in the database, if it exists.",
)
hash_index: QueryStr = Field(
None,
description="Tasks will be searched based on a hash of the defined Task. This is something which can "
"be generated by the Task spec itself and does not require server access to compute. "
"This should be unique in the database so there should be no reason to set anything else "
"if this is set as a query.",
)
program: QueryStr = Field(
None, description="Tasks will be searched based on the program responsible for executing this task."
)
status: QueryStr = Field(
driver: str
native: bool
queries: List[QueryData] = Field(
None,
description="List of queries to match against values columns. "
"See qcfractal.interface.collections.dataset_view.DatasetView.get_values",
)
subset: QueryStr
meta: EmptyMeta = Field(EmptyMeta(), description=common_docs[EmptyMeta])
data: Data = Field(..., description="Information about which values to return.")
class CollectionValueGETResponse(ProtoModel):
class Data(ProtoModel):
values: bytes = Field(..., description="Feather-serialized bytes representing a pandas DataFrame.")
units: Dict[str, str] = Field(..., description="Units of value columns.")
meta: CollectionSubresourceGETResponseMeta = Field(
..., description=str(get_base_docs(CollectionSubresourceGETResponseMeta))
)
data: Optional[Data] = Field(..., description="Values and units.")
register_model("collection/[0-9]+/value", "GET", CollectionValueGETBody, CollectionValueGETResponse)
class CollectionListGETBody(ProtoModel):
class Data(ProtoModel):
pass
@validator("priority", pre=True)
def munge_priority(cls, v):
if isinstance(v, str):
v = PriorityEnum[v.upper()]
return v
meta: Meta = Field(..., description="The additional specification information for the Task to add to the Database.")
data: List[Union[ObjectId, Molecule]] = Field(
...,
description="The list of either Molecule objects or Molecule Id's (those already in the database) to submit as "
"part of this Task.",
)
class TaskQueuePOSTResponse(ProtoModel):
meta: ResponsePOSTMeta = Field(..., description=common_docs[ResponsePOSTMeta])
data: ComputeResponse = Field(..., description="Data returned from the server from adding a Task.")
register_model("task_queue", "POST", TaskQueuePOSTBody, TaskQueuePOSTResponse)
class TaskQueuePUTBody(ProtoModel):
class Data(ProtoModel):
id: QueryObjectId = Field(
None,
description="The exact Id to target in database. If this is set as a search condition, there is no "
"reason to set anything else as this will be unique in the database, if it exists.",
)
base_result: QueryObjectId = Field( # TODO: Validate this description is correct
class BaseResultEnum(str, Enum):
result = "result"
procedure = "procedure"
class PythonComputeSpec(ProtoModel):
function: str = Field(
..., description="The module and function name of a Python-callable to call. Of the form 'module.function'."
)
args: List[Any] = Field(
..., description="A List of positional arguments to pass into ``function`` in order they appear."
)
kwargs: Dict[str, Any] = Field(..., description="Dictionary of keyword arguments to pass into ``function``.")
class TaskRecord(ProtoModel):
id: ObjectId = Field(None, description="The Database assigned Id of the Task, if it has been assigned yet.")
spec: PythonComputeSpec = Field(..., description="The Python function specification for this Task.")
parser: str = Field(..., description="The type of operation this is Task is. Can be 'single' or 'optimization'.")
status: TaskStatusEnum = Field(TaskStatusEnum.waiting, description="What stage of processing this task is at.")
# Compute blockers and prevention
program: str = Field(
..., description="Name of the quantum chemistry program which must be present to execute this task."
)
procedure: Optional[str] = Field(
None, description="Name of the procedure the compute platform must be able to perform to execute this task."
)
manager: Optional[str] = Field(None, description="The Queue Manager that evaluated this task.")
base_result: QueryStr = Field(
None,
description="The exact Id of the Result which this Task is linked to. If this is set as a "
"search condition, there is no reason to set anything else as this will be unique in the "
"database, if it exists. See also :class:`ResultRecord`.",
)
tag: QueryStr = Field(None, description="Tasks will be searched based on their associated tag.")
manager: QueryStr = Field(
None, description="Tasks will be searched based on the manager responsible for executing the task."
)
meta: QueryMetaFilter = Field(QueryMetaFilter(), description=common_docs[QueryMetaFilter])
data: Data = Field(..., description="The keys with data to search the database on for Tasks.")
class TaskQueueGETResponse(ProtoModel):
meta: ResponseGETMeta = Field(..., description=common_docs[ResponseGETMeta])
data: Union[List[TaskRecord], List[Dict[str, Any]]] = Field(
...,
description="Tasks found from the query. This is a list of :class:`TaskRecord` in most cases, however, "
"if a projection was specified in the GET request, then a dict is returned with mappings based "
"on the projection.",
)
register_model("task_queue", "GET", TaskQueueGETBody, TaskQueueGETResponse)
class TaskQueuePOSTBody(ProtoModel):
class Meta(ProtoModel):
procedure: str = Field(..., description="Name of the procedure which the Task will execute.")
program: str = Field(..., description="The program which this Task will execute.")
class TaskQueuePUTBody(ProtoModel):
class Data(ProtoModel):
id: QueryObjectId = Field(
None,
description="The exact Id to target in database. If this is set as a search condition, there is no "
"reason to set anything else as this will be unique in the database, if it exists.",
)
base_result: QueryObjectId = Field( # TODO: Validate this description is correct
None,
description="The exact Id of a result which this Task is slated to write to. If this is set as a "
"search condition, there is no reason to set anything else as this will be unique in the "
"database, if it exists. See also :class:`ResultRecord`.",
)
class Meta(ProtoModel):
operation: str = Field(..., description="The specific action you are taking as part of this update.")
@validator("operation")
def cast_to_lower(cls, v):
return v.lower()
meta: Meta = Field(..., description="The instructions to pass to the target Task from ``data``.")
data: Data = Field(..., description="The information which contains the Task target in the database.")
class TaskQueuePUTResponse(ProtoModel):
class Data(ProtoModel):
n_updated: int = Field(..., description="The number of tasks which were changed.")
meta: ResponseMeta = Field(..., description=common_docs[ResponseMeta])
data: Data = Field(..., description="Information returned from attempting updates of Tasks.")