|
2 | 2 | """ |
3 | 3 | Gemma python API (https://gemma.msl.ubc.ca/rest/v2/) |
4 | 4 | """ |
5 | | - |
6 | | -from gemmapy import sdk |
7 | | -from gemmapy import _processors as ps |
8 | | -from gemmapy import _validators as vs |
9 | | -from gemmapy import _subprocessors as sub |
| 5 | +import enum |
| 6 | +import json |
| 7 | +import logging |
| 8 | +import os |
| 9 | +import subprocess |
| 10 | +import warnings |
| 11 | +from getpass import getpass |
| 12 | +from io import StringIO |
10 | 13 | from typing import Optional, List, Callable |
11 | | -from pandas import DataFrame |
12 | | -import pandas as pd |
13 | | -import numpy as np |
| 14 | + |
14 | 15 | import anndata as ad |
| 16 | +import numpy as np |
| 17 | +import pandas as pd |
15 | 18 | from anndata import AnnData |
16 | | -from io import StringIO |
17 | | -import warnings |
18 | | -import json |
| 19 | +from pandas import DataFrame |
| 20 | + |
| 21 | +from gemmapy import _processors as ps |
| 22 | +from gemmapy import _subprocessors as sub |
| 23 | +from gemmapy import _validators as vs |
| 24 | +from gemmapy import sdk |
| 25 | + |
| 26 | +logger = logging.getLogger(__name__) |
19 | 27 |
|
| 28 | +class GemmaPath(enum.Enum): |
| 29 | + PROD = "prod" |
| 30 | + DEV = "dev" |
| 31 | + STAGING = "staging" |
20 | 32 |
|
21 | 33 | class GemmaPy(object): |
22 | 34 | """ |
23 | 35 | Main API class |
24 | 36 | """ |
25 | 37 |
|
26 | | - def __init__(self, auth:list|tuple=None, path="prod"): |
| 38 | + def __init__(self, auth: Optional[list | tuple] = None, |
| 39 | + path: Optional[GemmaPath | str] = None): |
27 | 40 | """ |
28 | 41 | :param list auth: (optional) A list or tuple of credential strings, e.g. |
29 | | - (your_username, your_password) |
30 | | - :param bool devel: (optional) If True development version of Gemma API will be |
31 | | - used. Default is False. |
| 42 | + (your_username, your_password). Note that you may also define your Gemma |
| 43 | + credentials using `GEMMA_USERNAME` and `GEMMA_PASSWORD` environment |
| 44 | + variables. For a more secure approach, you can also provide a |
| 45 | + `GEMMA_PASSWORD_CMD` variable that produces your password |
| 46 | + (e.g. `pass gemma` using https://www.passwordstore.org/). If only a |
| 47 | + username is supplied, a password prompt will be used. |
| 48 | + :param str path: (optional) Override the path to use for the REST API. |
| 49 | + You may use one of the enumerated values in GemmaPath or a string. |
| 50 | + Three special values are recognized: "prod", "staging" and "dev", |
| 51 | + although only "prod" is publicly accessible. The default is the value |
| 52 | + from the OpenAPI specification used to generate the SDK, which is |
| 53 | + usually equivalent to PROD. |
32 | 54 | """ |
33 | 55 |
|
34 | 56 | configuration = sdk.Configuration() |
35 | | - if path == "prod": |
36 | | - pass |
37 | | - # configuration.host = 'https://gemma.msl.ubc.ca/rest/v2' |
38 | | - elif path == 'dev': |
| 57 | + if path == GemmaPath.PROD or path == 'prod': |
| 58 | + logger.debug("Using production endpoint.") |
| 59 | + configuration.host = 'https://gemma.msl.ubc.ca/rest/v2' |
| 60 | + elif path == GemmaPath.DEV or path == 'dev': |
39 | 61 | configuration.host = 'https://dev.gemma.msl.ubc.ca/rest/v2' |
40 | | - elif path == 'staging': |
| 62 | + elif path == GemmaPath.STAGING or path == 'staging': |
41 | 63 | configuration.host = "https://staging-gemma.msl.ubc.ca/rest/v2" |
42 | | - else: |
| 64 | + elif path is not None: |
43 | 65 | configuration.host = path |
44 | | - |
| 66 | + else: |
| 67 | + # use the default configuration in the openapi.json file |
| 68 | + pass |
45 | 69 |
|
46 | 70 | if auth is not None: |
| 71 | + if len(auth) != 1 and len(auth) != 2: |
| 72 | + raise ValueError( |
| 73 | + 'There must be exactly one or two values in the auth parameter.') |
47 | 74 | configuration.username = auth[0] |
48 | | - configuration.password = auth[1] |
| 75 | + if len(auth) == 2: |
| 76 | + configuration.password = auth[1] |
| 77 | + else: |
| 78 | + configuration.password = getpass( |
| 79 | + f'Supply your password for {configuration.username}@{configuration.host}: ') |
| 80 | + elif os.environ.get('GEMMA_USERNAME'): |
| 81 | + logger.debug( |
| 82 | + 'Reading username for %s from $GEMMA_USERNAME.', |
| 83 | + configuration.host) |
| 84 | + configuration.username = os.getenv('GEMMA_USERNAME') |
| 85 | + if os.getenv('GEMMA_PASSWORD'): |
| 86 | + logger.debug("Reading password for %s@%s from $GEMMA_PASSWORD.", |
| 87 | + configuration.username, configuration.host) |
| 88 | + configuration.password = os.getenv('GEMMA_PASSWORD') |
| 89 | + elif os.getenv('GEMMA_PASSWORD_CMD'): |
| 90 | + logger.debug( |
| 91 | + "Reading password for %s@%s from $GEMMA_PASSWORD_CMD (%s).", |
| 92 | + configuration.username, configuration.host, |
| 93 | + os.getenv('GEMMA_PASSWORD_CMD')) |
| 94 | + password = subprocess.run(os.getenv('GEMMA_PASSWORD_CMD'), |
| 95 | + shell=True, check=True, |
| 96 | + stdout=subprocess.PIPE, |
| 97 | + text=True).stdout |
| 98 | + configuration.password = password.splitlines()[0] |
| 99 | + else: |
| 100 | + logger.debug( |
| 101 | + 'Could not read GEMMA_PASSWORD nor GEMMA_PASSWORD_CMD from environment, the password will be prompted.') |
| 102 | + configuration.password = getpass( |
| 103 | + f'Supply your password for {configuration.username}@{configuration.host}: ') |
49 | 104 |
|
50 | 105 | # create an instance of the API class |
51 | 106 | self.raw = sdk.DefaultApi(sdk.ApiClient(configuration)) |
|
0 commit comments