python_nedrex code (for now)

python_nedrex/python_nedrex/static.py

+from typing import Any as _Any
+from typing import Dict as _Dict
+from typing import Optional as _Optional
+
+from python_nedrex import config as _config
+from python_nedrex.common import check_response as _check_response
+from python_nedrex.common import download_file as _download_file
+from python_nedrex.common import http as _http
+
+
+def get_metadata() -> _Dict[str, _Any]:
+    url = f"{_config.url_base}/static/metadata"
+    resp = _http.get(url, headers={"x-api-key": _config.api_key})
+    result: _Dict[str, _Any] = _check_response(resp)
+    return result
+
+
+def get_license() -> str:
+    url = f"{_config.url_base}/static/license"
+    resp = _http.get(url)
+    result: str = _check_response(resp)
+    return result
+
+
+def download_lengths_map(target: _Optional[str] = None) -> None:
+    if target is None:
+        target = "lengths.map"
+
+    url = f"{_config.url_base}/static/lengths.map"
+
+    _download_file(url, target)

python_nedrex/python_nedrex/trustrank.py

+from typing import List as _List
+from typing import Optional as _Optional
+
+from python_nedrex import config as _config
+from python_nedrex.common import check_response as _check_response
+from python_nedrex.common import check_status_factory as _check_status_factory
+from python_nedrex.common import http as _http
+
+
+def trustrank_submit(
+    seeds: _List[str],
+    damping_factor: float = 0.85,
+    only_direct_drugs: bool = True,
+    only_approved_drugs: bool = True,
+    n: _Optional[int] = None,  # pylint: disable=C0103
+) -> str:
+    url = f"{_config.url_base}/trustrank/submit"
+
+    body = {
+        "seeds": seeds,
+        "damping_factor": damping_factor,
+        "only_direct_drugs": only_direct_drugs,
+        "only_approved_drugs": only_approved_drugs,
+        "N": n,
+    }
+
+    resp = _http.post(url, json=body, headers={"x-api-key": _config.api_key})
+    result: str = _check_response(resp)
+    return result
+
+
+check_trustrank_status = _check_status_factory("/trustrank/status")
+
+
+def download_trustrank_results(uid: str) -> str:
+    url = f"{_config.url_base}/trustrank/download"
+    params = {"uid": uid}
+
+    resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
+    result: str = _check_response(resp, return_type="text")
+    return result

python_nedrex/python_nedrex/validation.py

+from typing import List as _List
+
+from python_nedrex import config as _config
+from python_nedrex.common import check_response as _check_response
+from python_nedrex.common import check_status_factory as _check_status_factory
+from python_nedrex.common import http as _http
+
+check_validation_status = _check_status_factory("/validation/status")
+
+
+def check_module_member_type(mmt: str) -> None:
+    if mmt not in {"gene", "protein"}:
+        raise ValueError(f"module_member_type {mmt!r} is invalid (should be 'gene' or 'protein'")
+
+
+# pylint: disable=R0913
+def joint_validation_submit(
+    module_members: _List[str],
+    module_member_type: str,
+    test_drugs: _List[str],
+    true_drugs: _List[str],
+    permutations: int,
+    only_approved_drugs: bool = True,
+) -> str:
+    check_module_member_type(module_member_type)
+
+    url = f"{_config.url_base}/validation/joint"
+    body = {
+        "module_members": module_members,
+        "module_member_type": module_member_type,
+        "test_drugs": test_drugs,
+        "true_drugs": true_drugs,
+        "permutations": permutations,
+        "only_approved_drugs": only_approved_drugs,
+    }
+    resp = _http.post(url, json=body, headers={"x-api-key": _config.api_key})
+    result: str = _check_response(resp)
+    return result
+
+
+# pylint: enable=R0913
+
+
+def module_validation_submit(
+    module_members: _List[str],
+    module_member_type: str,
+    true_drugs: _List[str],
+    permutations: int,
+    only_approved_drugs: bool = True,
+) -> str:
+    check_module_member_type(module_member_type)
+
+    url = f"{_config.url_base}/validation/module"
+    body = {
+        "module_members": module_members,
+        "module_member_type": module_member_type,
+        "true_drugs": true_drugs,
+        "permutations": permutations,
+        "only_approved_drugs": only_approved_drugs,
+    }
+    resp = _http.post(url, json=body, headers={"x-api-key": _config.api_key})
+    result: str = _check_response(resp)
+    return result
+
+
+def drug_validation_submit(
+    test_drugs: _List[str], true_drugs: _List[str], permutations: int, only_approved_drugs: bool = True
+) -> str:
+
+    url = f"{_config.url_base}/validation/drug"
+    body = {
+        "test_drugs": test_drugs,
+        "true_drugs": true_drugs,
+        "permutations": permutations,
+        "only_approved_drugs": only_approved_drugs,
+    }
+
+    resp = _http.post(url, json=body, headers={"x-api-key": _config.api_key})
+    result: str = _check_response(resp)
+    return result

python_nedrex/python_nedrex/variants.py

+from typing import Any as _Any
+from typing import Dict as _Dict
+from typing import Generator as _Generator
+from typing import List as _List
+from typing import Optional as _Optional
+
+from python_nedrex import config as _config
+from python_nedrex.common import check_pagination_limit as _check_pagination_limit
+from python_nedrex.common import check_response as _check_response
+from python_nedrex.common import get_pagination_limit as _get_pagination_limit
+from python_nedrex.common import http as _http
+
+
+def get_effect_choices() -> _List[str]:
+    url = f"{_config.url_base}/variants/get_effect_choices"
+    resp = _http.get(url, headers={"x-api-key": _config.api_key})
+    result: _List[str] = _check_response(resp)
+    return result
+
+
+def get_review_status_choices() -> _List[str]:
+    url = f"{_config.url_base}/variants/get_review_choices"
+    resp = _http.get(url, headers={"x-api-key": _config.api_key})
+    result: _List[str] = _check_response(resp)
+    return result
+
+
+# pylint: disable=R0913
+def get_variant_disorder_associations(
+    variant_ids: _Optional[_List[str]] = None,
+    disorder_ids: _Optional[_List[str]] = None,
+    review_status: _Optional[_List[str]] = None,
+    effect: _Optional[_List[str]] = None,
+    limit: _Optional[int] = None,
+    offset: int = 0,
+) -> _List[_Dict[str, _Any]]:
+    max_limit = _get_pagination_limit()
+    if isinstance(limit, int):
+        _check_pagination_limit(limit, max_limit)
+    else:
+        limit = max_limit
+
+    params = {
+        "variant_id": variant_ids,
+        "disorder_id": disorder_ids,
+        "review_status": review_status,
+        "effect": effect,
+        "limit": limit,
+        "offset": offset,
+    }
+
+    url = f"{_config.url_base}/variants/get_variant_disorder_associations"
+    resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
+    result: _List[_Dict[str, _Any]] = _check_response(resp)
+    return result
+
+
+# pylint: enable=R0913
+
+
+def iter_variant_disorder_associations(
+    variant_ids: _Optional[_List[str]] = None,
+    disorder_ids: _Optional[_List[str]] = None,
+    review_status: _Optional[_List[str]] = None,
+    effect: _Optional[_List[str]] = None,
+) -> _Generator[_Dict[str, _Any], None, None]:
+    max_limit = _get_pagination_limit()
+    offset = 0
+
+    kwargs = {
+        "variant_ids": variant_ids,
+        "disorder_ids": disorder_ids,
+        "review_status": review_status,
+        "effect": effect,
+        "limit": max_limit,
+        "offset": offset,
+    }
+
+    while True:
+        results = get_variant_disorder_associations(**kwargs)
+        if len(results) == 0:
+            return
+        yield from results
+
+        offset += max_limit
+        kwargs["offset"] = offset
+
+
+def get_variant_gene_associations(
+    variant_ids: _Optional[_List[str]] = None,
+    gene_ids: _Optional[_List[str]] = None,
+    limit: _Optional[int] = None,
+    offset: int = 0,
+) -> _List[_Dict[str, _Any]]:
+
+    max_limit = _get_pagination_limit()
+    if isinstance(limit, int):
+        _check_pagination_limit(limit, max_limit)
+    else:
+        limit = max_limit
+
+    params = {
+        "variant_id": variant_ids,
+        "gene_id": gene_ids,
+        "offset": offset,
+        "limit": limit,
+    }
+
+    url = f"{_config.url_base}/variants/get_variant_gene_associations"
+    resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
+    result: _List[_Dict[str, _Any]] = _check_response(resp)
+    return result
+
+
+def iter_variant_gene_associations(
+    variant_ids: _Optional[_List[str]] = None,
+    gene_ids: _Optional[_List[str]] = None,
+) -> _Generator[_Dict[str, _Any], None, None]:
+    max_limit = _get_pagination_limit()
+    offset = 0
+
+    kwargs = {
+        "variant_ids": variant_ids,
+        "gene_ids": gene_ids,
+        "limit": max_limit,
+        "offset": offset,
+    }
+
+    while True:
+        results = get_variant_gene_associations(**kwargs)
+        if len(results) == 0:
+            return
+        yield from results
+
+        offset += max_limit
+        kwargs["offset"] = offset
+
+
+def get_variant_based_disorder_associated_genes(
+    disorder_id: str, review_status: _Optional[_List[str]] = None, effect: _Optional[_List[str]] = None
+) -> _List[_Dict[str, _Any]]:
+    params = {"disorder_id": disorder_id, "review_status": review_status, "effect": effect}
+
+    url = f"{_config.url_base}/variants/variant_based_disorder_associated_genes"
+
+    resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
+    result: _List[_Dict[str, _Any]] = _check_response(resp)
+    return result
+
+
+def get_variant_based_gene_associated_disorders(
+    gene_id: str, review_status: _Optional[_List[str]] = None, effect: _Optional[_List[str]] = None
+) -> _List[_Dict[str, _Any]]:
+    params = {"gene_id": gene_id, "review_status": review_status, "effect": effect}
+
+    url = f"{_config.url_base}/variants/variant_based_gene_associated_disorders"
+
+    resp = _http.get(url, params=params, headers={"x-api-key": _config.api_key})
+    result: _List[_Dict[str, _Any]] = _check_response(resp)
+    return result

python_nedrex/python_nedrex/vpd.py

+import os as _os
+from typing import Optional as _Optional
+
+from python_nedrex import config as _config
+from python_nedrex.common import http as _http
+from python_nedrex.decorators import check_url_vpd as _check_url_vpd
+
+
+@_check_url_vpd
+def get_vpd(disorder: str, number_of_patients: int, out_dir: str) -> _Optional[str]:
+    """
+    Downloads a .zip archive with the requested virtual patient data to the given directory.
+
+        Parameters:
+            disorder (str): The disorder mondoID (e.g mondo.0000090) for which the virtual patient should be retrieved.
+            number_of_patients (int): The number of simulated patients in the dataset. Can be 1, 10 or 100.
+            out_dir (str): The absolute path of a directory where the virtual patient data should be stored.
+
+        Returns:
+            archive (str): Absolute path of the downloaded zip archive or None if the requested resource does not exist.
+    """
+    archive_name: str = f"{disorder}_1000GP_{number_of_patients}VPSim.zip"
+    url: str = f"{_config.url_vpd}/vpd/{disorder}/{archive_name}"
+    archive: str = _os.path.join(out_dir, archive_name)
+
+    data = _http.get(url)
+    if data.status_code != 200:
+        return None
+
+    with open(archive, "wb") as arch:
+        arch.write(data.content)
+    return archive

python_nedrex/requirements.txt

+attrs==21.4.0
+cachetools==4.2.4
+more-itertools==8.13.0
+pytest==7.0.1  
+requests==2.27.1

python_nedrex/requirements_dev.txt

+pip==19.2.3
+bump2version==0.5.11
+wheel==0.33.6
+watchdog==0.9.0
+flake8==3.7.8
+tox==3.14.0
+coverage==4.5.4
+Sphinx==1.8.5
+twine==1.14.0
+
+pytest==6.2.4
+black==21.7b0

python_nedrex/setup.cfg

+[bumpversion]
+current_version = 0.1.1
+commit = True
+tag = True
+
+[bumpversion:file:setup.py]
+search = version='{current_version}'
+replace = version='{new_version}'
+
+[bumpversion:file:python_nedrex/__init__.py]
+search = __version__ = '{current_version}'
+replace = __version__ = '{new_version}'
+
+[bdist_wheel]
+universal = 1
+
+[flake8]
+exclude = docs
+[tool:pytest]
+

python_nedrex/setup.py

+#!/usr/bin/env python
+
+"""The setup script."""
+
+from setuptools import setup, find_packages
+
+with open("README.rst") as readme_file:
+    readme = readme_file.read()
+
+with open("HISTORY.rst") as history_file:
+    history = history_file.read()
+
+requirements = [
+    "attrs",
+    "requests",
+    "cachetools",
+]
+
+test_requirements = [
+    "pytest>=3",
+]
+
+setup(
+    author="David James Skelton",
+    author_email="james.skelton@newcastle.ac.uk",
+    python_requires=">=3.6",
+    classifiers=[
+        "Development Status :: 2 - Pre-Alpha",
+        "Intended Audience :: Developers",
+        "License :: OSI Approved :: MIT License",
+        "Natural Language :: English",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.6",
+        "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.8",
+    ],
+    description="A Python library for interfacing with the PNeDRex API",
+    install_requires=requirements,
+    license="MIT license",
+    long_description=readme + "\n\n" + history,
+    include_package_data=True,
+    keywords="python_nedrex",
+    name="python_nedrex",
+    packages=find_packages(include=["python_nedrex", "python_nedrex.*"]),
+    test_suite="tests",
+    tests_require=test_requirements,
+    url="https://github.com/james-skelton/python_nedrex",
+    version="0.1.1",
+    zip_safe=False,
+)

python_nedrex/tests/__init__.py

+"""Unit test package for python_nedrex."""

python_nedrex/tox.ini

+[tox]
+envlist = py36, py37, py38, flake8
+
+[travis]
+python =
+    3.8: py38
+    3.7: py37
+    3.6: py36
+
+[testenv:flake8]
+basepython = python
+deps = flake8
+commands = flake8 python_nedrex tests
+
+[testenv]
+setenv =
+    PYTHONPATH = {toxinidir}
+deps =
+    -r{toxinidir}/requirements_dev.txt
+; If you want to make tox run the tests with the same versions, create a
+; requirements.txt with the pinned versions and uncomment the following line:
+;     -r{toxinidir}/requirements.txt
+commands =
+    pip install -U pip
+    pytest --basetemp={envtmpdir}
+