# -*- coding: utf-8 -*-
"""Main code for interacting with APICURON."""
import datetime
from typing import Any, List, Mapping, Optional, Union
import pystow
import requests
from pydantic import BaseModel
__all__ = [
# URLs
"DESCRIPTION_URL",
"RESUBMISSION_URL",
# Data Models
"Term",
"Achievement",
"Description",
"Report",
"Submission",
# Utilities
"submit_description",
"resubmit_curations",
]
#: The endpoint for updating the description
DESCRIPTION_URL = "https://apicuron.bio.unipd.it/api/update_description"
#: An endpoint for total resubmission
RESUBMISSION_URL = "https://apicuron.org/api/resubmit_activity"
[docs]class Term(BaseModel):
"""A term."""
activity_term: str
activity_name: str
activity_category: str
description: str
score: int = 50
[docs]class Achievement(BaseModel):
"""An achievement."""
category: str # TODO this could be restricted with an enum
name: str
list_terms: List[str]
count_threshold: int = 10
color_code: str = "#055701"
[docs]class Description(BaseModel):
"""The description."""
resource_id: str
resource_name: str
resource_uri: str
resource_url: str
resource_long_name: str
resource_description: str
terms_def: List[Term]
achievements_def: List[Achievement]
[docs] def update_remote(self) -> requests.Response:
"""Update this description on the APICURON site."""
return requests.post(
DESCRIPTION_URL,
json=self.dict(),
headers=get_header(),
)
[docs]class Report(BaseModel):
"""A report on a single curation event."""
curator_orcid: str
resource_uri: str
#: Corresponds to the :`Term.activity_term` in an entry in `Description.terms_def` list
activity_term: str
timestamp: Optional[datetime.datetime] = None
[docs]class Submission(BaseModel):
"""A full submission."""
resource_uri: str
reports: List[Report]
time_start: Optional[datetime.datetime] = None
time_end: Optional[datetime.datetime] = None
[docs] def update_remote(self) -> requests.Response:
"""Update this resource on the APICURON site."""
return requests.post(
RESUBMISSION_URL,
json=self.dict(),
headers=get_header(),
)
def get_header():
"""Get the APICURON header information."""
token = pystow.get_config("apicuron", "token")
if token is None:
raise RuntimeError("missing APICURON_TOKEN")
header = {
"Authorization": f"Bearer {token}",
}
return header
[docs]def submit_description(payload: Union[Description, Mapping[str, Any]]) -> requests.Response:
"""Submit resource data."""
if not isinstance(payload, Description):
payload = Description(**payload)
return payload.update_remote()
[docs]def resubmit_curations(payload: Union[Submission, Mapping[str, Any]]) -> requests.Response:
"""Submit curations data."""
if not isinstance(payload, Submission):
payload = Submission(**payload)
return payload.update_remote()