Package phc

PHC SDK for Python

The phc-sdk-py is a developer kit for interfacing with the PHC API on Python 3.7 and above.

Project Status

GitHub PyPI status Downloads GitHub release Docs User Guides

Getting Started

Dependencies

Getting the Source

This project is hosted on GitHub.

Usage

A Session needs to be created first that stores the token and account information needed to access the PHC API. One can currently using API Key tokens generated from the PHC Account, or OAuth tokens generated using the CLI.

from phc import Session

session = Session(token=<TOKEN VALUE>, account="myaccount")

Once a Session is created, you can then access the different parts of the platform.

from phc.services import Accounts

accounts = Accounts(session)
myaccounts = accounts.get_list()

Contributing

We encourage public contributions! Please review CONTRIBUTING.md and CODE_OF_CONDUCT.md for details on our code of conduct and development process.

License

This project is licensed under the MIT License - see LICENSE file for details.

Authors

See the list of contributors who participate in this project.

Acknowledgements

This project is built with the following:

  • aiohttp - Asynchronous HTTP Client/Server for asyncio and Python.
Expand source code 
"""
.. include:: ../README.md
"""
import nest_asyncio
from phc.session import Session
from phc.api_response import ApiResponse
import phc.services as services
import phc.util as util

# https://markhneedham.com/blog/2019/05/10/jupyter-runtimeerror-this-event-loop-is-already-running/
nest_asyncio.apply()

__all__ = ["Session", "ApiResponse"]

__pdoc__ = {
    "version": False,
    "base_client": False,
    "api_response": False,
    "session": False,
}

Sub-modules

phc.adapter
phc.easy
phc.errors

A Python module for managing any client errors.

phc.services

Contains services for accessing different parts of the PHC platform.

phc.util

Module contains utility classes

Classes

class ApiResponse (*, client, http_verb: str, api_url: str, req_args: dict, data: Union[dict, str], headers: dict, status_code: int)

Represents an API response.

Attributes

nextPageToken : str
The nextPageToken for a paged response

Examples

>>> res = files.get_list(project_id="1234")
>>> print(str(res))
>>> print(res.nextPageToken)
Expand source code 
class ApiResponse:
    """Represents an API response.

    Attributes
    ----------
    nextPageToken : str
        The nextPageToken for a paged response

    Examples
    --------
    >>> res = files.get_list(project_id="1234")
    >>> print(str(res))
    >>> print(res.nextPageToken)
    """

    def __init__(
        self,
        *,
        client,
        http_verb: str,
        api_url: str,
        req_args: dict,
        data: Union[dict, str],
        headers: dict,
        status_code: int,
    ):
        self.http_verb = http_verb
        self.api_url = api_url
        self.req_args = req_args
        self.data = data
        self.headers = headers
        self.status_code = status_code
        self._initial_data = data
        self._client = client
        if isinstance(data, dict) and data.get("links", {}).get("next"):
            parsed = parse_qs(urlparse(data.get("links").get("next")).query)
            self.nextPageToken = parsed.get("nextPageToken")[0]

    def __str__(self):
        """Return the Response data if object is converted to a string."""
        return (
            json.dumps(self.data, indent=2)
            if isinstance(self.data, dict)
            else self.data
        )

    def __getitem__(self, key):
        """Retreives any key from the data store."""
        if isinstance(self.data, str):
            raise TypeError("Api response is text")

        return self.data.get(key, None)

    def get(self, key: str, default: Any = None):
        """Retreives any key from the response data.

        Parameters
        ----------
        key : str
            The key to fetch
        default : any, optional
            The default value to return if the key is not present, by default None

        Returns
        -------
        any
            The key value or the specified default if not present

        Raises
        ------
        TypeError
            If the api response is text
        """
        if isinstance(self.data, str):
            raise TypeError("Api response is text")

        return self.data.get(key, default)

    def get_as_dataframe(
        self, key: str, mapFunc: Optional[Callable[[Any], Any]] = None
    ):
        """Retrieves any key as a Panda DataFrame

        Parameters
        ----------
        key : str
            The key to fetch
        mapFunc : Callable[[Any], Any], optional
            A transform function to apply to each item before inserting into the DataFrame, by default None

        Returns
        -------
        DataFrame
            A Panda DataFrame

        Raises
        ------
        ImportError
            If pandas is not installed
        """
        if not _has_pandas:
            raise ImportError("pandas is required")

        if mapFunc is not None:
            mapped = list(map(mapFunc, self.data.get(key)))
            return _pd.DataFrame(mapped)

        # support OpenSearch sql response
        if key == "datarows" and self.data.get("schema") is not None:
            column_names = [col["name"] for col in self.data.get("schema")]
            return _pd.DataFrame(self.data.get(key), columns=column_names)

        return _pd.DataFrame(self.data.get(key))

    def validate(self):
        """Check if the response from API was successful.

        Returns
        -------
        ApiResponse
            This method returns its own object. e.g. 'self'

        Raises
        ------
        ApiError
            The request to the API failed.
        """
        if self.status_code >= 200 and self.status_code <= 300:
            return self
        msg = "The request to the API failed."
        raise e.ApiError(message=msg, response=self)

Methods

def get(self, key: str, default: Any = None)

Retreives any key from the response data.

Parameters

key : str
The key to fetch
default : any, optional
The default value to return if the key is not present, by default None

Returns

any
The key value or the specified default if not present

Raises

TypeError
If the api response is text
Expand source code 
def get(self, key: str, default: Any = None):
    """Retreives any key from the response data.

    Parameters
    ----------
    key : str
        The key to fetch
    default : any, optional
        The default value to return if the key is not present, by default None

    Returns
    -------
    any
        The key value or the specified default if not present

    Raises
    ------
    TypeError
        If the api response is text
    """
    if isinstance(self.data, str):
        raise TypeError("Api response is text")

    return self.data.get(key, default)
def get_as_dataframe(self, key: str, mapFunc: Optional[Callable[[Any], Any]] = None)

Retrieves any key as a Panda DataFrame

Parameters

key : str
The key to fetch
mapFunc : Callable[[Any], Any], optional
A transform function to apply to each item before inserting into the DataFrame, by default None

Returns

DataFrame
A Panda DataFrame

Raises

ImportError
If pandas is not installed
Expand source code 
def get_as_dataframe(
    self, key: str, mapFunc: Optional[Callable[[Any], Any]] = None
):
    """Retrieves any key as a Panda DataFrame

    Parameters
    ----------
    key : str
        The key to fetch
    mapFunc : Callable[[Any], Any], optional
        A transform function to apply to each item before inserting into the DataFrame, by default None

    Returns
    -------
    DataFrame
        A Panda DataFrame

    Raises
    ------
    ImportError
        If pandas is not installed
    """
    if not _has_pandas:
        raise ImportError("pandas is required")

    if mapFunc is not None:
        mapped = list(map(mapFunc, self.data.get(key)))
        return _pd.DataFrame(mapped)

    # support OpenSearch sql response
    if key == "datarows" and self.data.get("schema") is not None:
        column_names = [col["name"] for col in self.data.get("schema")]
        return _pd.DataFrame(self.data.get(key), columns=column_names)

    return _pd.DataFrame(self.data.get(key))
def validate(self)

Check if the response from API was successful.

Returns

ApiResponse
This method returns its own object. e.g. 'self'

Raises

ApiError
The request to the API failed.
Expand source code 
def validate(self):
    """Check if the response from API was successful.

    Returns
    -------
    ApiResponse
        This method returns its own object. e.g. 'self'

    Raises
    ------
    ApiError
        The request to the API failed.
    """
    if self.status_code >= 200 and self.status_code <= 300:
        return self
    msg = "The request to the API failed."
    raise e.ApiError(message=msg, response=self)
class Session (token: Optional[str] = None, refresh_token: Optional[str] = None, account: Optional[str] = None, adapter: Optional[Adapter] = None)

Represents a PHC API session

Initailizes a Session with token and account credentials.

Parameters

token : str, required
The PHC access token or API key, by default os.environ.get("PHC_ACCESS_TOKEN")
refresh_token : str, optional
The PHC refresh token, by default os.environ.get("PHC_REFRESH_TOKEN")
account : str, required
The PHC account ID, by default os.environ.get("PHC_ACCOUNT")
adapter : Adapter, optional
The adapter that executes requests
Expand source code 
class Session:
    """Represents a PHC API session"""

    adapter: Adapter

    def __init__(
        self,
        token: Optional[str] = None,
        refresh_token: Optional[str] = None,
        account: Optional[str] = None,
        adapter: Optional[Adapter] = None,
    ):
        """Initailizes a Session with token and account credentials.

        Parameters
        ----------
        token : str, required
            The PHC access token or API key, by default os.environ.get("PHC_ACCESS_TOKEN")

        refresh_token : str, optional
            The PHC refresh token, by default os.environ.get("PHC_REFRESH_TOKEN")

        account : str, required
            The PHC account ID, by default os.environ.get("PHC_ACCOUNT")

        adapter : Adapter, optional
            The adapter that executes requests
        """
        if not token:
            token = os.environ.get("PHC_ACCESS_TOKEN")

        if not refresh_token:
            refresh_token = os.environ.get("PHC_REFRESH_TOKEN")

        if not account:
            account = os.environ.get("PHC_ACCOUNT")

        if not adapter:
            adapter = Adapter()

        if adapter.should_refresh and (not token or not account):
            raise ValueError("Must provide a value for both token and account")

        self.token = token
        self.refresh_token = refresh_token
        self.account = account
        self.adapter = adapter

        hostname = urlparse(self._get_decoded_token().get("iss", "")).hostname
        env = (
            "dev"
            if hostname
            in ["cognito-idp.us-east-1.amazonaws.com", "api.dev.lifeomic.com"]
            else "us"
        )

        self.api_url = f"https://api.{env}.lifeomic.com/v1/"
        self.fhir_url = f"https://fhir.{env}.lifeomic.com/{account}/dstu3/"
        self.ga4gh_url = f"https://ga4gh.{env}.lifeomic.com/{account}/v1/"

    def _get_decoded_token(self):
        if self.token:
            return jwt.decode(
                self.token,
                options={"verify_signature": False},
                algorithms="RS256",
            )
        return {}

    def is_expired(self) -> bool:
        """Determines if the current access token is expired

        Returns
        -------
        bool
            True if there is no token or the token is expired, otherwise False
        """
        if self.adapter.should_refresh is False:
            return False

        return self._get_decoded_token().get("exp", 0) < time.time()

Class variables

var adapter : Adapter

Methods

def is_expired(self) ‑> bool

Determines if the current access token is expired

Returns

bool
True if there is no token or the token is expired, otherwise False
Expand source code 
def is_expired(self) -> bool:
    """Determines if the current access token is expired

    Returns
    -------
    bool
        True if there is no token or the token is expired, otherwise False
    """
    if self.adapter.should_refresh is False:
        return False

    return self._get_decoded_token().get("exp", 0) < time.time()