From 2b141234db0b12deda89f77bf2ce8792e0ce7de4 Mon Sep 17 00:00:00 2001
From: jwfraustro <36318163+jwfraustro@users.noreply.github.com>
Date: Thu, 30 Nov 2023 17:46:20 -0500
Subject: [PATCH] delete old models
---
tests/uws/__old__uws_models_test.py | 491 -------------------------
vo/models/xml/uws/__old__uws_models.py | 325 ----------------
vo/models/xml/uws/__old__uws_types.py | 34 --
3 files changed, 850 deletions(-)
delete mode 100644 tests/uws/__old__uws_models_test.py
delete mode 100644 vo/models/xml/uws/__old__uws_models.py
delete mode 100644 vo/models/xml/uws/__old__uws_types.py
diff --git a/tests/uws/__old__uws_models_test.py b/tests/uws/__old__uws_models_test.py
deleted file mode 100644
index 2a51e36..0000000
--- a/tests/uws/__old__uws_models_test.py
+++ /dev/null
@@ -1,491 +0,0 @@
-"""Tests for the XML serialization of UWS elements"""
-
-from datetime import timezone as tz
-from typing import Optional
-from unittest import TestCase
-
-from lxml import etree
-from pydantic_xml import element
-
-from asb.core.vo.uws.uws_models import (
- ErrorSummary,
- ExecutionPhase,
- Jobs,
- JobSummary,
- Parameter,
- Parameters,
- ResultReference,
- Results,
- ShortJobDescription,
-)
-from asb.core.vo.vo_models import VODateTime
-
-UWS_NAMESPACE_HEADER = """xmlns:uws="http://www.ivoa.net/xml/UWS/v1.0"
-xmlns:xlink="http://www.w3.org/1999/xlink"
-xmlns:xsd="http://www.w3.org/2001/XMLSchema"
-xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-"""
-
-# New schema versions can be downloaded from https://www.ivoa.net/xml/ under "UWS - Universal Worker Service"
-# The most current version is 1.1, found here: https://www.ivoa.net/xml/UWS/UWS-v1.1.xsd
-with open("asb/core/vo/uws/UWS-Schema-V1.0.xsd", "r") as schema_file:
- uws_schema = etree.XMLSchema(etree.parse(schema_file))
-
-
-class TestErrorSummaryType(TestCase):
- """Tests for the UWS errorSummary complex type"""
-
- test_error_summary_xml = f"""
- Invalid query.
-
- """
-
- def test_read_from_xml(self):
- """Test reading from XML"""
-
- error_summary = ErrorSummary.from_xml(self.test_error_summary_xml)
- self.assertEqual(error_summary.type, "transient")
- self.assertEqual(error_summary.has_detail, True)
- self.assertEqual(error_summary.message, "Invalid query.")
-
- def test_write_to_xml(self):
- """Test writing to XML"""
-
- error_summary = ErrorSummary(type="transient", has_detail=True, message="Invalid query.")
- error_summary_xml = error_summary.to_xml(encoding=str)
- self.assertIn('type="transient"', error_summary_xml)
- self.assertIn('hasDetail="true"', error_summary_xml)
- self.assertIn("Invalid query.", error_summary_xml)
-
-
-class TestParameterType(TestCase):
- """Tests for the UWS Parameter complex type"""
-
- test_parameter_xml = (
- f''
- "test_value"
- ""
- )
-
- def test_read_from_xml(self):
- """Test reading from XML"""
-
- parameter = Parameter.from_xml(self.test_parameter_xml)
- self.assertEqual(parameter.by_reference, False)
- self.assertEqual(parameter.id, "param1")
- self.assertEqual(parameter.is_post, False)
- self.assertEqual(parameter.value, "test_value")
-
- def test_write_to_xml(self):
- """Test writing to XML"""
-
- parameter = Parameter(by_reference=False, id="param1", is_post=False, value="test_value")
- parameter_xml = parameter.to_xml(encoding=str)
- self.assertIn('byReference="false"', parameter_xml)
- self.assertIn('id="param1"', parameter_xml)
- self.assertIn('isPost="false"', parameter_xml)
- self.assertIn("test_value", parameter_xml)
-
-
-class TestResultReferenceType(TestCase):
- """Test the UWS ResultReference complex type"""
-
- test_result_reference_xml = (
- f"'
- )
-
- def test_read_from_xml(self):
- """Test reading from XML"""
-
- result_reference = ResultReference.from_xml(self.test_result_reference_xml)
- self.assertEqual(result_reference.id, "result1")
- self.assertEqual(result_reference.mime_type, "text/xml")
- self.assertEqual(result_reference.href, "http://testlink.com/")
- self.assertEqual(result_reference.type, "simple")
- self.assertEqual(result_reference.size, 1234)
-
- def test_write_to_xml(self):
- """Test writing to XML"""
-
- result_reference = ResultReference(
- id="result1",
- mime_type="text/xml",
- href="http://testlink.com/",
- type="simple",
- size=1234,
- )
- result_reference_xml = result_reference.to_xml(encoding=str)
- self.assertIn('id="result1"', result_reference_xml)
- self.assertIn('mime-type="text/xml"', result_reference_xml)
- self.assertIn('xlink:href="http://testlink.com/"', result_reference_xml)
- self.assertIn('xlink:type="simple"', result_reference_xml)
- self.assertIn('size="1234"', result_reference_xml)
-
-
-class TestResultsElement(TestCase):
- """Test the results list element"""
-
- test_results_xml = (
- f""
- ""
- ""
- ""
- )
-
- def test_read_from_xml(self):
- """Test reading from XML"""
-
- results = Results.from_xml(self.test_results_xml)
- self.assertEqual(len(results.results), 2)
- self.assertEqual(results.results[0].id, "result1")
- self.assertEqual(results.results[1].id, "result2")
-
- def test_write_to_xml(self):
- """Test writing to XML"""
-
- results_list = Results(
- results=[
- ResultReference(
- id="result1",
- mime_type="text/xml",
- href="http://testlink.com/",
- ),
- ResultReference(
- id="result2",
- mime_type="text/xml",
- href="http://testlink.com/",
- ),
- ]
- )
- results_xml = results_list.to_xml(encoding=str, skip_empty=True)
- self.assertIn('id="result1"', results_xml)
- self.assertIn('id="result2"', results_xml)
-
- def test_validate(self):
- """Test validation against XML schema"""
-
- results = Results(
- results=[
- ResultReference(
- id="result1",
- mime_type="text/xml",
- href="http://testlink.com/",
- ),
- ResultReference(
- id="result2",
- mime_type="text/xml",
- href="http://testlink.com/",
- ),
- ]
- )
- results_xml = etree.fromstring(results.to_xml(encoding=str, skip_empty=True))
- uws_schema.assertValid(results_xml)
-
-
-class TestShortJobDescriptionType(TestCase):
- """Test the UWS ShortJobDescription complex type"""
-
- test_short_job_description_xml = (
- f''
- "PENDING"
- "runId1"
- "1900-01-01T01:01:01Z"
- ""
- )
-
- def test_read_from_xml(self):
- """Test reading from XML"""
-
- short_job_description = ShortJobDescription.from_xml(self.test_short_job_description_xml)
- self.assertEqual(short_job_description.job_id, "id1")
- self.assertEqual(short_job_description.type, "simple")
- self.assertEqual(short_job_description.href, "http://uri1")
- self.assertEqual(short_job_description.phase, "PENDING")
- self.assertEqual(short_job_description.run_id, "runId1")
- self.assertEqual(short_job_description.owner_id.value, None)
- self.assertEqual(short_job_description.creation_time, VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc))
-
- def test_write_to_xml(self):
- """Test writing to XML"""
-
- short_job_description = ShortJobDescription(
- job_id="id1",
- owner_id=None,
- type="simple",
- href="http://uri1",
- phase="PENDING",
- run_id="runId1",
- creation_time=VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc),
- )
- short_job_description_xml = short_job_description.to_xml(skip_empty=True, encoding=str)
- self.assertIn('id="id1"', short_job_description_xml)
- self.assertIn('', short_job_description_xml)
- self.assertIn('xlink:type="simple"', short_job_description_xml)
- self.assertIn('xlink:href="http://uri1"', short_job_description_xml)
- self.assertIn("PENDING", short_job_description_xml)
- self.assertIn("runId1", short_job_description_xml)
- self.assertIn("1900-01-01T01:01:01.000Z", short_job_description_xml)
-
-
-class TestGenericParametersElement(TestCase):
- """Test the generic UWS Parameters element"""
-
- class TestParameters(Parameters):
- """A test generic Parameters class"""
-
- param1: Optional[Parameter] = element(tag="parameter")
- param2: Optional[Parameter] = element(tag="parameter")
- param3: Optional[Parameter] = element(tag="parameter")
-
- test_parameters_xml = (
- f""
- 'value1'
- 'value2'
- 'value3'
- ""
- )
-
- def test_base_parameters(self):
- """Test the base Parameters element - no subclass"""
-
- parameters = Parameters(
- param1=Parameter(id="param1", type="xs:string", value="value1"),
- param2=Parameter(id="param2", type="xs:string", value="value2"),
- param3=Parameter(id="param3", type="xs:string", value="value3"),
- )
- # pylint: disable=no-member
- self.assertEqual(parameters.param1.value, "value1")
- self.assertEqual(parameters.param2.value, "value2")
- self.assertEqual(parameters.param3.value, "value3")
-
- def test_read_from_dict(self):
- """Test reading base Parameters from dict"""
-
- # Occurs when reading from cache
- param_dict = {
- "param1": {"id": "param1", "type": "xs:string", "value": "value1"},
- "param2": {"id": "param2", "type": "xs:string", "value": "value2"},
- "param3": {"id": "param3", "type": "xs:string", "value": "value3"},
- }
- # pylint: disable=no-member
- parameters = Parameters(**param_dict)
- self.assertEqual(parameters.param1.value, "value1")
- self.assertEqual(parameters.param2.value, "value2")
- self.assertEqual(parameters.param3.value, "value3")
-
- def test_read_from_xml(self):
- """Test reading from XML"""
-
- parameters = self.TestParameters.from_xml(self.test_parameters_xml)
- self.assertEqual(len(parameters.dict()), 3)
-
- self.assertEqual(parameters.param1.id, "param1")
- self.assertEqual(parameters.param2.id, "param2")
- self.assertEqual(parameters.param3.id, "param3")
-
- self.assertEqual(parameters.param1.value, "value1")
- self.assertEqual(parameters.param2.value, "value2")
- self.assertEqual(parameters.param3.value, "value3")
-
- def test_write_to_xml(self):
- """Test writing to XML"""
-
- parameters = self.TestParameters(
- param1=Parameter(id="param1", type="xs:string", value="value1"),
- param2=Parameter(id="param2", type="xs:string", value="value2"),
- param3=Parameter(id="param3", type="xs:string", value="value3"),
- )
- parameters_xml = parameters.to_xml(skip_empty=True, encoding=str)
- self.assertIn('id="param1"', parameters_xml)
- self.assertIn('id="param2"', parameters_xml)
- self.assertIn('id="param3"', parameters_xml)
-
- self.assertIn("value1", parameters_xml)
- self.assertIn("value2", parameters_xml)
- self.assertIn("value3", parameters_xml)
-
- def test_validate(self):
- """Test validation against XML schema"""
-
- parameters = self.TestParameters(
- param1=Parameter(id="param1", type="xs:string", value="value1"),
- param2=Parameter(id="param2", type="xs:string", value="value2"),
- param3=Parameter(id="param3", type="xs:string", value="value3"),
- )
- parameters_xml = etree.fromstring(parameters.to_xml(skip_empty=True, encoding=str))
- uws_schema.assertValid(parameters_xml)
-
-
-class TestJobSummaryElement(TestCase):
- """Test the UWS JobSummary element"""
-
- class TestParameters(Parameters):
- """A test generic Parameters class"""
-
- param1: Optional[Parameter] = element(tag="parameter")
- param2: Optional[Parameter] = element(tag="parameter")
-
- job_summary_xml = (
- f''
- "jobId1"
- "runId1"
- "ownerId1"
- "PENDING"
- ''
- "1900-01-01T01:01:01.000Z"
- "1900-01-01T01:01:01.000Z"
- "1900-01-01T01:01:01.000Z"
- "0"
- "1900-01-01T01:01:01.000Z"
- ""
- 'value1'
- 'value2'
- ""
- ""
- ""
- "jobInfo1"
- ""
- )
-
- def test_read_from_xml(self):
- """Test reading from XML"""
-
- job_summary = JobSummary[self.TestParameters].from_xml(self.job_summary_xml)
- self.assertEqual(job_summary.job_id, "jobId1")
- self.assertEqual(job_summary.run_id, "runId1")
- self.assertEqual(job_summary.owner_id.value, "ownerId1")
- self.assertEqual(job_summary.phase, ExecutionPhase.PENDING.value)
- self.assertEqual(job_summary.quote.value, None)
- self.assertEqual(job_summary.creation_time, VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc))
- self.assertEqual(job_summary.start_time.value, VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc).isoformat())
- self.assertEqual(job_summary.end_time.value, VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc).isoformat())
- self.assertEqual(job_summary.execution_duration, 0)
- self.assertEqual(job_summary.destruction, VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc))
- self.assertEqual(len(job_summary.parameters.dict()), 2)
- self.assertEqual(job_summary.parameters.param1.id, "param1")
- self.assertEqual(job_summary.parameters.param2.id, "param2")
- self.assertEqual(job_summary.parameters.param1.value, "value1")
- self.assertEqual(job_summary.parameters.param2.value, "value2")
- self.assertEqual(len(job_summary.results.results), 0)
- self.assertEqual(job_summary.error_summary.message, "")
- self.assertEqual(job_summary.job_info[0], "jobInfo1")
-
- def test_write_to_xml(self):
- """Test writing to XML"""
-
- job_summary = JobSummary[self.TestParameters](
- job_id="jobId1",
- run_id="runId1",
- owner_id="ownerId1",
- phase=ExecutionPhase.PENDING,
- quote=None,
- creation_time=VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc),
- start_time=None,
- end_time=None,
- execution_duration=0,
- destruction=VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc),
- parameters=self.TestParameters(
- param1=Parameter(id="param1", type="xs:string", value="value1"),
- param2=Parameter(id="param2", type="xs:string", value="value2"),
- ),
- results=Results(),
- error_summary=ErrorSummary(),
- job_info=["jobInfo1"],
- )
- job_summary_xml = job_summary.to_xml(skip_empty=True, encoding=str)
- self.assertIn("jobId1", job_summary_xml)
- self.assertIn("runId1", job_summary_xml)
- self.assertIn("ownerId1", job_summary_xml)
- self.assertIn("PENDING", job_summary_xml)
- self.assertIn("1900-01-01T01:01:01.000Z", job_summary_xml)
- self.assertIn("1900-01-01T01:01:01.000Z", job_summary_xml)
- self.assertIn("1900-01-01T01:01:01.000Z", job_summary_xml)
- self.assertIn("1900-01-01T01:01:01.000Z", job_summary_xml)
- self.assertIn("value1", job_summary_xml)
- self.assertIn("value2", job_summary_xml)
- self.assertIn("jobInfo1", job_summary_xml)
-
- def test_validate(self):
- """Validate against the schema"""
-
- job_summary = JobSummary[self.TestParameters](
- job_id="jobId1",
- run_id="runId1",
- owner_id="ownerId1",
- phase=ExecutionPhase.PENDING,
- quote=None,
- creation_time=VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc),
- start_time=None,
- end_time=None,
- execution_duration=0,
- destruction=VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc),
- parameters=self.TestParameters(
- param1=Parameter(id="param1", value="value1"),
- param2=Parameter(id="param2", value="value2"),
- ),
- results=Results(results=[ResultReference(id="result1")]),
- error_summary=None,
- )
- job_summary_xml = etree.fromstring(job_summary.to_xml(skip_empty=True, encoding=str))
- uws_schema.assertValid(job_summary_xml)
-
-
-class TestJobsElement(TestCase):
- """Test the UWS Jobs element"""
-
- test_job_list_xml = (
- f''
- ''
- "PENDING"
- ''
- "1900-01-01T01:01:01Z"
- ""
- ""
- )
-
- def test_read_from_xml(self):
- """Test reading from XML"""
-
- jobs_element = Jobs.from_xml(self.test_job_list_xml)
- self.assertEqual(len(jobs_element.jobref), 1)
- self.assertEqual(jobs_element.jobref[0].job_id, "id1")
- self.assertEqual(jobs_element.jobref[0].owner_id.value, None)
- self.assertEqual(jobs_element.jobref[0].phase, ExecutionPhase.PENDING)
- self.assertEqual(jobs_element.jobref[0].creation_time, VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc))
-
- def test_write_to_xml(self):
- """Test writing to XML"""
-
- jobs_element = Jobs(
- jobref=[
- ShortJobDescription(
- job_id="id1",
- phase=ExecutionPhase.PENDING,
- creation_time=VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc),
- )
- ]
- )
- jobs_element_xml = jobs_element.to_xml(skip_empty=True, encoding=str)
- self.assertIn("id1", jobs_element_xml)
- self.assertIn("PENDING", jobs_element_xml)
- self.assertIn("1900-01-01T01:01:01.000Z", jobs_element_xml)
-
- def test_validate(self):
- """Validate against the schema"""
-
- jobs_element = Jobs(
- jobref=[
- ShortJobDescription(
- job_id="id1",
- phase=ExecutionPhase.PENDING,
- creation_time=VODateTime(1900, 1, 1, 1, 1, 1, tzinfo=tz.utc),
- )
- ]
- )
- jobs_element_xml = etree.fromstring(jobs_element.to_xml(skip_empty=True, encoding=str))
- uws_schema.assertValid(jobs_element_xml)
diff --git a/vo/models/xml/uws/__old__uws_models.py b/vo/models/xml/uws/__old__uws_models.py
deleted file mode 100644
index aa052bb..0000000
--- a/vo/models/xml/uws/__old__uws_models.py
+++ /dev/null
@@ -1,325 +0,0 @@
-"""UWS Job Schema using Pydantic-XML models"""
-from datetime import timedelta, timezone
-from typing import Any, Generic, Optional, TypeVar, Union
-
-from pydantic import validator
-from pydantic_xml import BaseXmlModel, attr, element
-
-from vo.models.xml.generics import VODateTime
-from vo.models.xml.uws.__old__uws_types import ErrorTypeName, ExecutionPhase, UWSVersions
-from vo.models.xml.xlink import TypeValue
-
-# pylint: disable=line-too-long
-# pylint: disable=no-self-argument
-
-NSMAP = {
- "uws": "http://www.ivoa.net/xml/UWS/v1.0",
- "xlink": "http://www.w3.org/1999/xlink",
- "xsd": "http://www.w3.org/2001/XMLSchema",
- "xsi": "http://www.w3.org/2001/XMLSchema-instance",
-}
-
-RESULT_EXPIRATION_SEC = 24 * 60 * 60 # 1 day in seconds
-
-
-def validate_nillable(value, expected_type) -> dict:
- """Creates a dictionary representing a NillableElement for a pydantic-xml model.
-
- The dictionary creates the element by way of NillableElement(**dict) in pydantic-xml internally.
- It also handles deserialized cases, when reading from the cache, for example.
-
- Example:
- the element
-
- - job_summary.start_time = 01-01-1970
- -- pydantic-xml receives NillableElement(**{value:'01-01-1970', nil:None})
- -- the element is represented as 01-01-1970
-
- - job_summary.start_time = None (or value was not provided)
- -- pydantic-xml receives NillableElement(**{value:None, nil:'true'})
- -- the element is represented as
-
- When serialized (not XML - in the cache as a dict) the element is represented by:
- if defined:
- start_time = {'value':'01-01-1970', 'nil'=None}
- or if not defined:
- start_time = None
-
- When JobSummary elements are deserialized from the cache, these elements are handled similarly
- as the above cases.
- """
- if value:
- if isinstance(value, NillableElement):
- return value
- if isinstance(value, dict):
- # if we received a dict to validate, we're reading this value from the cache, and was
- # already coerced into the format of a NillableElement dict.
- if value.get("value"):
- # if value was previously set, check it (mostly in cases of bad cache/model data)
- if not isinstance(value.get("value"), expected_type):
- raise ValueError(f"Incorrect type for value: {value}. Expected type {str(expected_type)}")
- return value
-
- if not isinstance(value, expected_type):
- # if we actually got a value, it's probably from user or code input - validate it
- raise ValueError(f"Incorrect type for value: {value}. Expected type {str(expected_type)}")
- return {"value": value}
-
- return {"nil": "true"}
-
-
-class NillableElement(BaseXmlModel, ns="uws", nsmap=NSMAP):
- """A generic used for elements that may have a value, but must display a 'xsi:nil' attribute when they do not.
-
- Used to maintain compatibility with the pydantic-xml 'to_xml()' function parameter 'skip_empty=True'
-
- The skip_empty=True parameter allows XML elements with no content (optionals) to be skipped when serializing
- elements. Without it, pydantic-xml will attempt to serialize every element and attribute. Optionals with None,
- will fail and throw an error. Therefore, if we have an element that can be None, but must also be displayed
- and have a 'nil' attribute, we need to handle it differently.
-
- The 'validate_nillable' function above will create a blank nillable element for a job attribute that was not
- defined (owner_id=None) but must be displayed. Those elements are ownerId, quote, startTime, endTime and destruction.
- We define destruction at job creation, so it is not necessary to check.
- """
-
- value: Optional[str]
- nil: Optional[str] = attr(name="nil", ns="xsi")
-
-
-class DatetimeElement(BaseXmlModel, ns="uws", nsmap=NSMAP):
- """A wrapper element for a VODatetime object.
-
- Necessary to allow a Union between a NillableElement and simple VODatetime datatype.
- """
-
- __root__: VODateTime
-
-
-class ErrorSummary(BaseXmlModel, tag="errorSummary", ns="uws", nsmap=NSMAP):
- """ErrorSummary element - a short summary of a UWS Job error
-
- Included as part of a JobSummary - a full representation of the error is available
- from the /{job_id}/error endpoint.
-
- Elements:
- message: a brief summary of the error.
- Attributes:
- type: characterization of the error type - transient or fatal
- has_detail: if true there is a more detailed error message at the errors endpoint
-
- """
-
- message: str = element(tag="message", default="")
-
- type: ErrorTypeName = attr(name="type", default=ErrorTypeName.TRANSIENT)
- has_detail: bool = attr(name="hasDetail", default=False)
-
-
-class Parameter(BaseXmlModel, tag="parameter", ns="uws", nsmap=NSMAP):
- """Parameter element - list of input parameters to an async job
-
- value: the text content of the parameter tag
-
- Attributes
- id: the parameter's identifier e.g. 'maxrec', 'catalog'
- byReference: if true, the content of the parameter is a URL where the actual parameter value is stored
- is_post: isn't documented in the UWS spec...
- """
-
- value: Optional[str]
- by_reference: Optional[bool] = attr(name="byReference")
- id: str = attr(name="id")
- is_post: Optional[bool] = attr(name="isPost")
-
-
-class ResultReference(BaseXmlModel, tag="result", ns="uws", nsmap=NSMAP):
- """ResultReference element - simple container for xlink result references"""
-
- id: str = attr(name="id", default="result")
- type: Optional[TypeValue] = attr(
- name="type",
- ns="xlink",
- default=TypeValue.SIMPLE,
- )
- href: Optional[str] = attr(ns="xlink")
- size: Optional[int] = attr()
- mime_type: Optional[str] = attr(name="mime-type")
-
-
-class ShortJobDescription(BaseXmlModel, tag="jobref", ns="uws", nsmap=NSMAP):
- """ShortJobDescription element - a brief description of a UWS Job
-
- This is returned as part of a JobList query, to keep the list of results brief.
-
- Elements:
- phase: the execution phase - returned at /{job_id}/phase
- runId: a client-supplied identifier - UWS simply returns it when referencing the job
- ownerId: the creator of the job - a parsable string, otherwise a nill element
- creationTime: the instance a job was created
- Attributes:
- job_id: UUID4 identifier for the job
- xsi:type/href: a link to the full job summary endpoint
- """
-
- phase: ExecutionPhase = element(tag="phase")
- run_id: Optional[str] = element(tag="runId")
- owner_id: Optional[NillableElement] = element(tag="ownerId")
- creation_time: VODateTime = element(tag="creationTime")
-
- job_id: str = attr(name="id")
- type: Optional[TypeValue] = attr(ns="xlink", name="type", default=TypeValue.SIMPLE)
- href: Optional[str] = attr(ns="xlink")
-
- @validator("owner_id", pre=True, always=True)
- def validate_ownerid(cls, value):
- """Validate the owner_id - create a nillable element if it doesn't exist"""
- return validate_nillable(value, str)
-
-
-class Parameters(BaseXmlModel, tag="parameters", ns="uws", nsmap=NSMAP):
- """A generic UWS parameters type
-
- Used to create a list of parameters for a UWS JobSummary object.
- Parameters subtypes should be created for specific UWS applications with the appropriate
- service-specific parameters. See asb.core.vo.uws.tap_models.TAPParameters for an example.
- """
-
- def __init__(__pydantic_self__, **data: Any) -> None:
- # during init -- especially if reading from xml -- we may not get the parameters in the order
- # pydantic-xml expects. This will remap the dict with keys based on the parameter id.
- parameter_vals = [val for val in data.values() if val is not None]
- remapped_vals = {}
- for param in parameter_vals:
- if isinstance(param, dict):
- remapped_vals[param["id"]] = Parameter(**param)
- else:
- remapped_vals[param.id] = param
- data = remapped_vals
- super().__init__(**data)
-
- class Config:
- """UWS Parameters config"""
-
- extra = "allow"
-
-
-# pylint: disable=invalid-name
-ParametersType = TypeVar("ParametersType", bound=Parameters)
-
-
-class Results(BaseXmlModel, tag="results", ns="uws", nsmap=NSMAP):
- """Results element - a simple container holding a list of Result elements"""
-
- results: list[ResultReference] = element(tag="result", default_factory=lambda: [])
-
-
-class JobSummary(BaseGenericXmlModel, Generic[ParametersType], tag="job", ns="uws", nsmap=NSMAP):
- """JobSummary element - The complete representation of the state of a UWS Job
-
- JobSummary takes a Generic type for the Parameters element, allowing for service-specific
- parameters (TAP for example) to be included in the JobSummary. This allows us to use the common model structure
- of the UWS JobSummary with any parameters, but swap in alternate parameters when stricter
- service-specific validation is needed.
-
- For pydantic docs on generic models see: https://docs.pydantic.dev/latest/usage/models/#generic-models
- And for generic xml models with pydantic-xml: https://pydantic-xml.readthedocs.io/en/latest/pages/data-binding/generics.html
-
- Attributes:
- version (schema_version) - the UWS service version. Required by UWS V1.1
- Version 1.0 is included in an Enum for read-support only.
- Elements:
- job_id: the server-provided UUID4 identifier for the job
- run_id: a client-provided string - UWS simply returns it as part of the job
- owner_id: the creator of the job - a parsable string, otherwise a nill element
- phase: the execution phase of the job - returned at /{job_id}/phase
- quote: when the job is likely to complete (datetime) or a nill value
- -- not supported in our implementation
- creation_time: the datetime when the job was created
- start_time: the datetime a job started execution
- end_time: the datetime a job finished execution
- execution_duration: how long (in seconds) a job is permitted to run - 0 for unlimited
- destruction: the datetime at which the job + records + results will be destroyed
- parameters: a Parameters object containing a list of job creation parameters
- results: the results for the job - also retrieved at /{job_id}/results
- error_summary: a brief description of an error, if one occurred during the job
- job_info: arbitrary information that can be provided by the UWS service
- """
-
- schema_version: UWSVersions = attr(name="version", default=UWSVersions.V1_1.value)
-
- job_id: str = element(tag="jobId")
- run_id: Optional[str] = element(tag="runId")
- owner_id: Optional[NillableElement] = element(tag="ownerId")
- phase: Optional[ExecutionPhase] = element(tag="phase", default=ExecutionPhase.PENDING)
- quote: Optional[NillableElement] = element(tag="quote")
- creation_time: Optional[VODateTime] = element(
- tag="creationTime", default_factory=lambda: VODateTime.now(timezone.utc)
- )
- start_time: Optional[Union[NillableElement, DatetimeElement]] = element(tag="startTime")
- end_time: Optional[Union[NillableElement, DatetimeElement]] = element(tag="endTime")
- execution_duration: Optional[int] = element(tag="executionDuration", default=0)
- destruction: Optional[VODateTime] = element(
- tag="destruction",
- default_factory=lambda: VODateTime.now(timezone.utc) + timedelta(seconds=RESULT_EXPIRATION_SEC),
- )
- parameters: ParametersType = element(tag="parameters", default_factory=lambda: [])
- results: Optional[Results] = element(tag="results")
- error_summary: Optional[ErrorSummary] = element(tag="errorSummary")
- job_info: Optional[list[str]] = element(tag="jobInfo")
-
- @validator("run_id")
- def validate_runid_length(cls, value):
- """Validate the run_id is < 64 characters. Should also be handled by the FastAPI endpoint."""
- if value:
- if len(value) > 64:
- raise ValueError("runID value must be less than 64 characters")
- return value
-
- @validator("owner_id", pre=True, always=True)
- def validate_ownerid(cls, value):
- """Set the owner_id element if one was provided, otherwise set the element to nil"""
- return validate_nillable(value, str)
-
- @validator("quote", pre=True, always=True)
- def validate_quote(cls, value):
- """Set the quote element if one was provided, otherwise set the element to nil"""
- return validate_nillable(value, str)
-
- @validator("start_time", pre=True, always=True)
- def validate_starttime(cls, value):
- """Set the start_time element to nil at creation -
- it is expected to be there even if the job has not yet started"""
- # the VODateTime object is serialized/deserialized as its ISO formatted string
- return validate_nillable(value, str)
-
- @validator("end_time", pre=True, always=True)
- def validate_endtime(cls, value):
- """Set the end_time element to nil at creation -
- it is expected to be there even if the job has not yet started"""
- # the VODateTime object is serialized/deserialized as its ISO formatted string
- return validate_nillable(value, str)
-
-
-class Job(BaseXmlModel, tag="job", ns="uws", nsmap=NSMAP):
- """The root element for a JobSummary response.
-
- Used to wrap the JobSummary element in an element of type <{uws}:job>
- """
-
- __root__: JobSummary
-
-
-class Jobs(BaseXmlModel, tag="jobs", ns="uws", nsmap=NSMAP):
- """The root element for a Job List response
-
- Used to wrap a list of ShortJobDescription elements in a <{uws}:jobs> tag
-
- Attributes:
- version: the UWS service version
- Elements:
- The list of ShortJobDescription elements, each having <{uws}:jobref> tag
- """
-
- version: UWSVersions = attr(default=UWSVersions.V1_1)
- jobref: list[ShortJobDescription] = element(tag="jobref", default_factory=lambda: [])
diff --git a/vo/models/xml/uws/__old__uws_types.py b/vo/models/xml/uws/__old__uws_types.py
deleted file mode 100644
index f42b6c8..0000000
--- a/vo/models/xml/uws/__old__uws_types.py
+++ /dev/null
@@ -1,34 +0,0 @@
-"""Simple types for UWS models"""
-from enum import Enum
-
-
-class ExecutionPhase(str, Enum):
- """Enum for valid async job phases. Not necessarily all supported.
- From https://www.ivoa.net/documents/UWS/20161024/REC-UWS-1.1-20161024.html#ExecutionPhase"""
-
- PENDING = "PENDING"
- QUEUED = "QUEUED"
- EXECUTING = "EXECUTING"
- COMPLETED = "COMPLETED"
- ERROR = "ERROR"
- ABORTED = "ABORTED"
- UNKNOWN = "UNKNOWN"
- HELD = "HELD"
- SUSPENDED = "SUSPENDED"
- ARCHIVED = "ARCHIVED"
- RUN = "RUN" # nonstandard support for astropy/TAPPlus
-
-
-class ErrorTypeName(str, Enum):
- """Enum for error types in job summary.
- From https://www.ivoa.net/documents/UWS/20161024/REC-UWS-1.1-20161024.html#Error"""
-
- TRANSIENT = "transient"
- FATAL = "fatal"
-
-
-class UWSVersions(str, Enum):
- """Supported UWS version numbers"""
-
- V1_0 = "1.0"
- V1_1 = "1.1"