client constructors are `api` and `api_from_config`.
"""
-from __future__ import absolute_import
-from future import standard_library
-standard_library.install_aliases()
-from builtins import range
import collections
-import http.client
import httplib2
import json
import logging
import time
import types
+from typing import (
+ Any,
+ Dict,
+ List,
+ Mapping,
+ Optional,
+)
+
import apiclient
import apiclient.http
from apiclient import discovery as apiclient_discovery
errors.ApiError, *args, **kwargs)
apiclient_errors.HttpError.__new__ = staticmethod(_new_http_error)
-def http_cache(data_type):
+def http_cache(data_type: str) -> cache.SafeHTTPCache:
"""Set up an HTTP file cache
This function constructs and returns an `arvados.cache.SafeHTTPCache`
- backed by the filesystem under `~/.cache/arvados/`, or `None` if the
- directory cannot be set up. The return value can be passed to
+ backed by the filesystem under a cache directory from the environment, or
+ `None` if the directory cannot be set up. The return value can be passed to
`httplib2.Http` as the `cache` argument.
Arguments:
- * data_type: str --- The name of the subdirectory under `~/.cache/arvados`
+ * data_type: str --- The name of the subdirectory
where data is cached.
"""
try:
- homedir = pathlib.Path.home()
- except RuntimeError:
- return None
- path = pathlib.Path(homedir, '.cache', 'arvados', data_type)
- try:
- path.mkdir(parents=True, exist_ok=True)
- except OSError:
+ path = util._BaseDirectories('CACHE').storage_path() / data_type
+ path.mkdir(exist_ok=True)
+ except (OSError, RuntimeError):
return None
- return cache.SafeHTTPCache(str(path), max_age=60*60*24*2)
+ else:
+ return cache.SafeHTTPCache(str(path), max_age=60*60*24*2)
def api_client(
- version,
- discoveryServiceUrl,
- token,
+ version: str,
+ discoveryServiceUrl: str,
+ token: str,
*,
- cache=True,
- http=None,
- insecure=False,
- num_retries=10,
- request_id=None,
- timeout=5*60,
- **kwargs,
-):
+ cache: bool=True,
+ http: Optional[httplib2.Http]=None,
+ insecure: bool=False,
+ num_retries: int=10,
+ request_id: Optional[str]=None,
+ timeout: int=5*60,
+ **kwargs: Any,
+) -> apiclient_discovery.Resource:
"""Build an Arvados API client
This function returns a `googleapiclient.discovery.Resource` object
Keyword-only arguments:
* cache: bool --- If true, loads the API discovery document from, or
- saves it to, a cache on disk (located at
- `~/.cache/arvados/discovery`).
+ saves it to, a cache on disk.
* http: httplib2.Http | None --- The HTTP client object the API client
object will use to make requests. If not provided, this function will
Additional keyword arguments will be passed directly to
`googleapiclient.discovery.build`.
-
"""
if http is None:
http = httplib2.Http(
return svc
def normalize_api_kwargs(
- version=None,
- discoveryServiceUrl=None,
- host=None,
- token=None,
- **kwargs,
-):
+ version: Optional[str]=None,
+ discoveryServiceUrl: Optional[str]=None,
+ host: Optional[str]=None,
+ token: Optional[str]=None,
+ **kwargs: Any,
+) -> Dict[str, Any]:
"""Validate kwargs from `api` and build kwargs for `api_client`
This method takes high-level keyword arguments passed to the `api`
**kwargs,
}
-def api_kwargs_from_config(version=None, apiconfig=None, **kwargs):
+def api_kwargs_from_config(
+ version: Optional[str]=None,
+ apiconfig: Optional[Mapping[str, str]]=None,
+ **kwargs: Any
+) -> Dict[str, Any]:
"""Build `api_client` keyword arguments from configuration
This function accepts a mapping with Arvados configuration settings like
**kwargs,
)
-def api(version=None, cache=True, host=None, token=None, insecure=False,
- request_id=None, timeout=5*60, *,
- discoveryServiceUrl=None, **kwargs):
+def api(
+ version: Optional[str]=None,
+ cache: bool=True,
+ host: Optional[str]=None,
+ token: Optional[str]=None,
+ insecure: bool=False,
+ request_id: Optional[str]=None,
+ timeout: int=5*60,
+ *,
+ discoveryServiceUrl: Optional[str]=None,
+ **kwargs: Any,
+) -> 'arvados.safeapi.ThreadSafeApiCache':
"""Dynamically build an Arvados API client
This function provides a high-level "do what I mean" interface to build an
from .safeapi import ThreadSafeApiCache
return ThreadSafeApiCache({}, {}, kwargs, version)
-def api_from_config(version=None, apiconfig=None, **kwargs):
+def api_from_config(
+ version: Optional[str]=None,
+ apiconfig: Optional[Mapping[str, str]]=None,
+ **kwargs: Any
+) -> 'arvados.safeapi.ThreadSafeApiCache':
"""Build an Arvados API client from a configuration mapping
This function builds an Arvados API client from a mapping with user