"""API for the sbmlreport web service.
This provides basic functionality of
parsing the model and returning the JSON representation based on fastAPI.
"""
import tempfile
import time
import traceback
import uuid
from pathlib import Path
from typing import Any, Dict, Optional, Union
import requests
import uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from pymetadata.core.annotation import RDFAnnotation, RDFAnnotationData
from pymetadata.identifiers.miriam import BQB
from pymetadata.omex import EntryFormat, Manifest, ManifestEntry, Omex
from sbmlutils import log
from sbmlutils.console import console
from sbmlutils.report.api_examples import ExampleMetaData, examples_info
from sbmlutils.report.sbmlinfo import SBMLDocumentInfo
[docs]logger = log.get_logger(__name__)
[docs]api = FastAPI(
title="sbml4humans",
description="sbml4humans backend api",
version="0.1.2",
terms_of_service="https://github.com/matthiaskoenig/sbmlutils/blob/develop/sbml4humans/privacy_notice.md",
contact={
"name": "Matthias König",
"url": "https://livermetabolism.com",
"email": "konigmatt@googlemail.com",
},
license_info={
"name": "LGPLv3",
"url": "http://opensource.org/licenses/LGPL-3.0",
},
openapi_tags=[
{
"name": "examples",
"description": "Manage and query examples.",
},
{
"name": "metadata",
"description": "Manage and query metadata.",
},
{
"name": "reports",
"description": "Create report data.",
},
],
)
# API Permissions Data
[docs]origins = [
# "localhost",
# "localhost:3456",
# "sbml4humans.de",
"*"
]
api.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@api.get("/api/examples", tags=["examples"])
[docs]def examples() -> Dict[Any, Any]:
"""Get examples for reports."""
try:
return {"examples": [v.dict() for v in examples_info.values()]}
except Exception as e:
return _handle_error(e)
@api.get("/api/examples/{example_id}", tags=["examples"])
[docs]def example(example_id: str) -> Dict[Any, Any]:
"""Get specific example."""
try:
example: Optional[ExampleMetaData] = examples_info.get(example_id, None)
content: Dict
if example:
source: Path = example.file # type: ignore
content = json_for_omex(omex_path=source)
else:
content = {"error": f"example for id does not exist '{example_id}'"}
return content
except Exception as e:
return _handle_error(e)
@api.post("/api/file", tags=["reports"])
[docs]async def report_from_file(request: Request) -> Dict[Any, Any]:
"""Upload file and return JSON report."""
try:
file_data = await request.form()
file_content = await file_data["source"].read() # type: ignore
# write in file; check if SBML; gzip SBML or OMEX,
# TODO: implement `is_omex` function
with tempfile.TemporaryDirectory() as tmp_dir:
path = Path(tmp_dir) / "file"
with open(path, "w") as f:
if isinstance(file_content, bytes):
file_content = file_content.decode("utf-8")
f.write(file_content)
return json_for_omex(path)
except Exception as e:
return _handle_error(e, info={})
@api.get("/api/url", tags=["reports"])
[docs]def report_from_url(url: str) -> Dict[Any, Any]:
"""Get JSON report via URL."""
try:
response = requests.get(url)
response.raise_for_status()
with tempfile.TemporaryDirectory() as f_tmp:
path = Path(f_tmp) / "file"
with open(path, "w") as f:
f.write(response.text)
return json_for_omex(path)
except Exception as e:
return _handle_error(e, info={"url": url})
@api.post("/api/content", tags=["reports"])
[docs]async def get_report_from_content(request: Request) -> Dict[Any, Any]:
"""Get JSON report from file contents."""
file_content: Optional[str] = None
try:
file_content_bytes: bytes = await request.body()
file_content = file_content_bytes.decode("utf-8")
with tempfile.TemporaryDirectory() as f_tmp:
path = Path(f_tmp) / "file"
with open(path, "w") as f:
f.write(file_content)
return json_for_omex(path)
except Exception as e:
return _handle_error(e, info={})
[docs]def json_for_omex(omex_path: Path) -> Dict[str, Any]:
"""Create json for omex path.
Path can be either Omex or an SBML file.
"""
uid: str = uuid.uuid4().hex
if Omex.is_omex(omex_path):
omex = Omex().from_omex(omex_path)
else:
# Path is SBML we create a new archive
omex = Omex()
omex.add_entry(
entry_path=omex_path,
entry=ManifestEntry(
location="./model.xml", format=EntryFormat.SBML, master=True
),
)
content = {"uid": uid, "manifest": omex.manifest.dict(), "reports": {}}
# Add report JSON for all SBML files
entry: ManifestEntry
for entry in omex.manifest.entries:
if entry.is_sbml():
sbml_path: Path = omex.get_path(entry.location)
content["reports"][entry.location] = json_for_sbml( # type: ignore
uid=uid, source=sbml_path
)
return content
[docs]def json_for_sbml(uid: str, source: Union[Path, str, bytes]) -> Dict:
"""Create JSON content for given SBML source.
Source is either path to SBML file or SBML string.
"""
if isinstance(source, bytes):
source = source.decode("utf-8")
time_start = time.time()
info = SBMLDocumentInfo.from_sbml(source=source)
time_elapsed = round(time.time() - time_start, 3)
debug = False
if debug:
console.rule("Creating JSON content")
console.print(info.info)
console.rule()
logger.info(f"JSON created for '{uid}' in '{time_elapsed}'")
return {
"report": info.info,
"debug": {
"jsonReportTime": f"{time_elapsed} [s]",
},
}
[docs]def _handle_error(e: Exception, info: Optional[Dict] = None) -> Dict[Any, Any]:
"""Handle exceptions in the backend.
All calls are wrapped in a try/except which will provide the errors to the frontend.
:param info: optional dictionary with information.
"""
res = {
"errors": [
f"{e.__str__()}",
f"{''.join(traceback.format_exception(None, e, e.__traceback__))}",
],
"warnings": [],
"info": info,
}
return res
@api.get("/api/annotation_resource", tags=["metadata"])
[docs]def get_annotation_resource(resource: str) -> Dict[Any, Any]:
"""Get information for annotation_resource.
Used to resolve annotation information.
:param resource: unique identifier of resource (url or miriam urn)
:return: Response
"""
try:
annotation = RDFAnnotation(qualifier=BQB.IS, resource=resource)
data = RDFAnnotationData(annotation=annotation)
return data.to_dict() # type: ignore
except Exception as e:
return _handle_error(e)
if __name__ == "__main__":
# shell command: uvicorn sbmlutils.report.api:app --reload --port 1444
uvicorn.run(
"sbmlutils.report.api:api",
host="localhost",
port=1444,
log_level="info",
reload=True,
)