diff --git a/Dockerfile b/Dockerfile index 26262a1..aa00aa8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,6 +2,6 @@ FROM python:3.6 WORKDIR /code COPY . /code/ -RUN pip3 install --no-cache-dir flake8 +RUN pip3 install --no-cache-dir flake8 responses RUN python setup.py install ENTRYPOINT ["python"] diff --git a/README.md b/README.md index 6a5a5d6..56bf274 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,14 @@ Docker image with pre-installed library is also available, run it via: docker run -i -t quay.io/keboola/sapi-python-client ``` +## Tests + +```bash +$ git clone https://github.com/keboola/sapi-python-client.git && cd sapi-python-client +$ python setup.py test +``` + + Under development -- all contributions very welcome :) Kickstarted via https://gist.github.com/Halama/6006960 diff --git a/kbcstorage/base.py b/kbcstorage/base.py new file mode 100644 index 0000000..0829176 --- /dev/null +++ b/kbcstorage/base.py @@ -0,0 +1,128 @@ +""" +Base classes for constructing the client. + +Primarily exposes a base Endpoint class which deduplicates functionality across +various endpoints, such as tables, workspaces, jobs, etc. as described in the +`Storage API documentation`. + + +.. _Storage API documentation: + http://docs.keboola.apiary.io/ +""" +import requests + + +class Endpoint: + """ + Base class for implementing a single endpoint related to a single entities + as described in the Storage API. + + Attributes: + base_url (str): The base URL for this endpoint. + token (str): A key for the Storage API. + """ + def __init__(self, root_url, path_component, token): + """ + Create an endpoint. + + Args + root_url (str): Root url of API. eg. + "https://connection.keboola.com/v2/storage/" + path_component (str): The section of the path specific to the + endpoint. eg. "buckets" + token (str): A key for the Storage API. Can be found in the storage + console. + """ + self.base_url = '{}/{}'.format(root_url.strip('/'), + path_component.strip('/')) + self.token = token + + def get(self, *args, **kwargs): + """ + Construct a requests GET call with args and kwargs and process the + results. + + Args: + *args: Positional arguments to pass to the get request. + **kwargs: Key word arguments to pass to the get request. + + Returns: + body: Response body parsed from json. + + Raises: + requests.HTTPError: If the API request fails. + """ + r = requests.get(*args, **kwargs) + try: + r.raise_for_status() + except requests.HTTPError: + # Handle different error codes + raise + else: + return r.json() + + def post(self, *args, **kwargs): + """ + Construct a requests POST call with args and kwargs and process the + results. + + Args: + *args: Positional arguments to pass to the post request. + **kwargs: Key word arguments to pass to the post request. + + Returns: + body: Response body parsed from json. + + Raises: + requests.HTTPError: If the API request fails. + """ + r = requests.post(*args, **kwargs) + try: + r.raise_for_status() + except requests.HTTPError: + # Handle different error codes + raise + else: + return r.json() + + def put(self): + """ + **Not implemented** + + Construct a requests PUT call with args and kwargs and process the + result + + Args: + *args: Positional arguments to pass to the put request. + **kwargs: Key word arguments to pass to the put request. + + Returns: + body: Response body parsed from json. + + Raises: + requests.HTTPError: If the API request fails. + """ + raise NotImplementedError + + def delete(self, *args, **kwargs): + """ + Construct a requests DELETE call with args and kwargs and process the + result + + Args: + *args: Positional arguments to pass to the delete request. + **kwargs: Key word arguments to pass to the delete request. + + Returns: + body: Response body parsed from json. + + Raises: + requests.HTTPError: If the API request fails. + """ + r = requests.delete(*args, **kwargs) + try: + r.raise_for_status() + except requests.HTTPError: + # Handle different error codes + raise + # Should delete return something on success? diff --git a/kbcstorage/buckets.py b/kbcstorage/buckets.py new file mode 100644 index 0000000..20adafe --- /dev/null +++ b/kbcstorage/buckets.py @@ -0,0 +1,143 @@ +""" +Manages calls to the Storage API relating to buckets + +Full documentation `here`. + +.. _here: + http://docs.keboola.apiary.io/#reference/buckets/ +""" +from kbcstorage.base import Endpoint + + +class Buckets(Endpoint): + """ + Buckets Endpoint + """ + def __init__(self, root_url, token): + """ + Create a Workspaces endpoint. + + Args: + url (:obj:`str`): The base url for the API. + token (:obj:`str`): A storage API key. + """ + super().__init__(root_url, 'buckets', token) + + def list(self): + """ + Get all job details. + + Returns: + response_body: The parsed json from the HTTP response. + + Raises: + requests.HTTPError: If the API request fails. + """ + headers = {'X-StorageApi-Token': self.token} + + return self.get(self.base_url, headers=headers) + + def detail(self, bucket_id): + """ + Retrieves information about a given bucket. + + Args: + bucket_id (int or str): The id of the bucket. + + Raises: + requests.HTTPError: If the API request fails. + """ + url = '{}/{}'.format(self.base_url, bucket_id) + headers = {'X-StorageApi-Token': self.token} + + return self.get(url, headers=headers) + + def create(self, name, stage='in', description='', backend=None): + """ + Create a new bucket. + + Args: + name (str): The new bucket name (only alphanumeric and underscores) + stage (str): The new bucket stage. Can be one of ``in`` or ``out``. + Default ``in``. + description (str): The new bucket description. + backend (str): The new bucket backend. Cand be one of + ``snowflake``, ``redshift`` or ``mysql``. Default determined by + project settings. + Returns: + response_body: The parsed json from the HTTP response. + + Raises: + requests.HTTPError: If the API request fails. + """ + # Separating create and link into two distinct functions... + headers = { + 'X-StorageApi-Token': self.token, + 'Content-Type': 'application/x-www-form-urlencoded' + } + # Need to check args... + body = { + 'name': name, + 'stage': stage, + 'description': description, + 'backend': backend + } + + return self.post(self.base_url, headers=headers, data=body) + + def delete(self, bucket_id, force=False): + """ + Delete a bucket referenced by ``bucket_id``. + + By default, only empty buckets without dependencies (aliases etc) can + be deleted. The optional ``force`` parameter allows for the deletion + of non-empty buckets. + + Args: + bucket_id (str or int): The id of the bucket to be deleted. + force (bool): If ``True``, deletes the bucket even if it is not + empty. Default ``False``. + """ + # How does the API handle it when force == False and the bucket is non- + # empty? + url = '{}/{}'.format(self.base_url, bucket_id) + headers = {'X-StorageApi-Token': self.token} + params = {'force': force} + super().delete(url, headers=headers, params=params) + + def link(self, *args, **kwargs): + """ + **Not implemented** + + Link an existing bucket from another project. + + Creates a new bucket which contains the contents of a shared bucket in + a source project. Linking a bucket from another project is only + possible if it has been enabled in the project. + """ + raise NotImplementedError + + def share(self, *args, **kwargs): + """ + **Not implemented** + + Enable sharing of a bucket. + + The bucket will be shared to the entire organisation to which the + project belongs. It may then be shared to any project of that + organization. This operation is only available to administrator tokens. + """ + raise NotImplementedError + + def unshare(self, *args, **kwargs): + """ + **Not implemented** + + Stop sharing a bucket. + + The bucket must not be linked to other projects. To unshare an already + linked bucket, the links must first be deleted - use ``delete`` on the + bucket in the linking project. This operation is only available for + administrator tokens. + """ + raise NotImplementedError diff --git a/kbcstorage/client.py b/kbcstorage/client.py index 21ad633..6050d03 100644 --- a/kbcstorage/client.py +++ b/kbcstorage/client.py @@ -1,228 +1,33 @@ -import boto3 -from botocore.exceptions import ClientError -import os -import requests -import string -import time +""" +Entry point for the Storage API client. +""" +from kbcstorage.buckets import Buckets +from kbcstorage.workspaces import Workspaces +from kbcstorage.jobs import Jobs -class HttpHelper: - api_url = "https://connection.keboola.com" - api_version = 'v2' - user_agent = "Keboola StorageApi Python Client/v2" - - def __init__(self, token, api_url=None): - self.token = token - if api_url is not None: - self.api_url = api_url - - def tokenheader(self): - return { - 'User-Agent': self.user_agent, - 'X-StorageApi-Token': self.token - } - - def getRequest(self, url, params = None): - if params == None: - params = [] - resp = self.getAbsUrlRequest(url, params) - resp.raise_for_status() - return resp.json() - - def postRequest(self, url, params): - resp = requests.post(self.api_url + '/' + self.api_version + '/' + url, headers=self.tokenheader(), data=params) - resp.raise_for_status() - return resp.json() - - def getAbsUrlRequest(self, url, params): - return requests.get(self.api_url + '/' + self.api_version + '/' + url, headers=self.tokenheader(), params=params) - - def deleteRequest(self, url, params): - resp = requests.delete(self.api_url + '/' + self.api_version + '/' + url, headers=self.tokenheader(), data=params) - resp.raise_for_status() class Client: - def __init__(self, token, api_url=None): + """ + Storage API Client. + """ + def __init__(self, api_domain, token): + """ + Initialise a client. + + Args: + api_domain (str): The domain on which the API sits. eg. + "https://connection.keboola.com". + token (str): A storage API key. + """ + api_version_string = 'v2' + api_path_component = 'storage' + + self.root_url = '{}/{}/{}'.format(api_domain, + api_version_string, + api_path_component) self.token = token - self.http = HttpHelper(token, api_url) - - def list_buckets(self): - return self.http.getRequest('storage/buckets') - - def list_bucket_tables(self, bucket): - return self.http.getRequest('storage/buckets/' + bucket + '/tables') - - def files_prepare(self, name, sizeBytes): - payload = {'name': name, 'sizeBytes': sizeBytes, 'notify': False} - return self.http.postRequest('storage/files/prepare', payload) - - def files_upload(self, path): - fileObject = open(path, 'rb') - fileResource = self.files_prepare(os.path.basename(path), os.path.getsize(path)) - uploadParams = fileResource['uploadParams'] - params = { - 'key': uploadParams['key'], - 'acl': uploadParams['acl'], - 'signature': uploadParams['signature'], - 'policy': uploadParams['policy'], - 'AWSAccessKeyId': uploadParams['AWSAccessKeyId'] - } - files = {'file': fileObject} - requests.post(uploadParams['url'], data=params, files=files) - return fileResource - - def bucket_exists(self, bucket_id): - try: - self.http.getRequest('storage/buckets/' + bucket_id) - except requests.exceptions.HTTPError as e: - if e.response.status_code == 404: - return False - else: - raise - return True - - def table_exists(self, tableId): - try: - resp = self.http.getRequest('storage/tables/' + tableId) - return True - except requests.exceptions.HTTPError as e: - return False - - def get_table(self, tableId): - return self.http.getRequest('storage/tables/' + tableId) - - def get_bucket(self, bucketId): - return self.http.getRequest('storage/buckets/' + bucketId) - - def load_table_async(self, tableId, options = None): - opts = self.prepare_options(options) - if "federationToken" not in opts: - opts["federationToken"] = 1 - return self.http.postRequest("storage/tables/" + tableId + "/export-async", opts) - - def get_job_status(self, url): - resp = self.http.getAbsUrlRequest(url, []) - return resp.json() - - def get_file_info(self, fileId, federationToken = "1"): - return self.http.getRequest("storage/files/" + str(fileId), {"federationToken": federationToken}) - - def prepare_options(self, options = None): - if options is None: - options = {} - # which parameters are allowed - params = ["limit", "changedSince", "changedUntil", "whereColumn", "whereValues"] - opts = {} - for val in params: - if val in options: - opts[val] = options[val] - - if "columns" in options: - opts["columns"] = string.split(options["columns"], ",") - - if "whereValues" in options: - for val in options["whereValues"]: - opts["whereValies[" + val + "]"] = options["whereValues"][val] - #opts[[paste0("whereValues[", i - 1, "]")]] < - options[["whereValues"]][i] - - return opts - - def get_table_data(self, tableId, localFile, options = None): - resp = self.load_table_async(tableId, options) - retries = 1 - while True: - job = self.get_job_status(resp["url"]) - if job["status"] == "success": - break - time.sleep(2 ^ retries) - retries = retries + 1 - if job["status"] != "waiting" and job["status"] != "processing": - raise Exception("Job status: " + job["status"] + " - " + job["error"]["message"] + " {" + job["error"]["exceptionId"] + ")") - - table = self.get_table(tableId) - fileInfo = self.get_file_info(job["results"]["file"]["id"]) - s3 = boto3.resource( - 's3', - aws_access_key_id=fileInfo["credentials"]["AccessKeyId"], - aws_secret_access_key=fileInfo["credentials"]["SecretAccessKey"], - aws_session_token=fileInfo["credentials"]["SessionToken"] - ) - - if fileInfo["isSliced"]: - manifest = self.http.getAbsUrlRequest(fileInfo["url"], []).json() - fileNames = [] - for entry in manifest["entries"]: - fullPath = entry["url"] - fileName = fullPath.rsplit("/", 1)[1] - fileNames.append(fileName) - splittedPath = string.split(fullPath, "/") - fileKey = "/".join(splittedPath[3:]) - bucket = s3.Bucket(fileInfo["s3Path"]["bucket"]) - try: - bucket.download_file(fileKey, fileName) - except ClientError as e: - if e.response['Error']['Code'] == "404": - print("Could not find the object in s3") - else: - raise - - # merge the downloaded files - with open(localFile, 'w') as outfile: - for fileName in fileNames: - with open(fileName) as infile: - for line in infile: - outfile.write(line) - os.remove(fileName) - - else: - # single file is friendlier - bucket = s3.Bucket(fileInfo["s3Path"]["bucket"]) - bucket.download_file(fileInfo["s3Path"]["key"], localFile) - - - def save_table(self, tableName, bucket, localFilePath, options = None): - - if options is None: - options = {} - - tableId = bucket + "." + tableName - postUrl = "storage/buckets/" + bucket + "/tables-async" - if self.table_exists(tableId): - postUrl = "storage/tables/" + tableId + "/import-async" - - resource = self.files_upload(localFilePath) - opts = { - "bucketId": bucket, - "name": tableName, - "dataFileId": resource["id"] - } - opts['delimeter'] = options["delimeter"] if "delimeter" in options else "," - opts['enclosure'] = options["enclosure"] if "enclosure" in options else '"' - opts['escapedBy'] = options["escapedBy"] if "escapedBy" in options else None - opts['primaryKey'] = options["primaryKey"] if "primaryKey" in options else None - opts['incremental'] = options["incremental"] if "incremental" in options else None - - jobres = self.http.postRequest(postUrl, opts) - - retries = 1 - while True: - job = self.get_job_status(jobres["url"]) - if job["status"] == "success": - break - time.sleep(2 ^ retries) - retries = retries + 1 - if job["status"] != "waiting" and job["status"] != "processing": - raise Exception( - "Job status: " + job["status"] + " - " + job["error"]["message"] - + " {" + job["error"]["exceptionId"] + ")" - ) - return True - - def drop_bucket(self, bucket_id, options=None): - return self.http.deleteRequest('storage/buckets/' + str(bucket_id), options) - def create_bucket(self, name, stage, description='', backend=None): - data = {'name': name, 'stage': stage, 'description': description} - if backend is not None: - data['backend'] = backend - return self.http.postRequest('storage/buckets/', data)['id'] + self.buckets = Buckets(self.root_url, self.token) + self.workspaces = Workspaces(self.root_url, self.token) + self.jobs = Jobs(self.root_url, self.token) diff --git a/kbcstorage/jobs.py b/kbcstorage/jobs.py new file mode 100644 index 0000000..e12e666 --- /dev/null +++ b/kbcstorage/jobs.py @@ -0,0 +1,138 @@ +""" +Manages calls to the Storage API relating to jobs. + +Full documentation `here`. + +.. _here: + http://docs.keboola.apiary.io/#reference/jobs/ +""" +import time + +from kbcstorage.base import Endpoint + + +class Jobs(Endpoint): + """ + Jobs are objects that manage asynchronous tasks, these are all + potentially long-running actions such as loading table data, + snapshotting, table structure modifications. Jobs are created by + actions on target resources. + + A job has four available statuses: + + ``waiting`` + The job is in the queue and is waiting for execution. + + ``processing`` + The job is being processed by a worker. + + ``success`` + The job is done with a success. + + ``error`` + The job is done with an error. + """ + + def __init__(self, root_url, token): + """ + Create a Jobs endpoint. + + Args: + url (:obj:`str`): The base url for the API. + token (:obj:`str`): A storage API key. + """ + super().__init__(root_url, 'jobs', token) + + def list(self): + """ + List all jobs details. + + Returns: + response_body: The json from the HTTP response. + + Raises: + requests.HTTPError: If the API request fails. + """ + headers = {'X-StorageApi-Token': self.token} + + return self.get(self.base_url, headers=headers) + + def detail(self, job_id): + """ + Retrieves information about a given job. + + Args: + job_id (int or str): The id of the job. + + Raises: + requests.HTTPError: If the API request fails. + """ + headers = {'X-StorageApi-Token': self.token} + url = '{}/{}'.format(self.base_url, job_id) + + return self.get(url, headers=headers) + + def status(self, job_id): + """ + Retrieve the status of a given job. + + Args: + job_id (int or str): The id of the job. + + Raises: + requests.HTTPError: If the API request fails. + """ + return self.detail(job_id)['status'] + + def completed(self, job_id): + """ + Check if a job is completed or not. + + Args: + job_id (int or str): The id of the job. + + Returns: + completed (bool): True if job is completed, else False. + + Raises: + requests.HTTPError: If the API request fails. + """ + completed_statuses = ('error', 'success') + return self.status(job_id) in completed_statuses + + def block_until_completed(self, job_id, d=1): + """ + Poll the API until the job is completed. + + Args: + job_id (int or str): The id of the job + d (int): The time delta between successive polls in seconds. + Default 1. + + Raises: + requests.HTTPError: If any API request fails. + """ + while not self.completed(job_id): + time.sleep(d) + + def block_for_success(self, job_id, d=1): + """ + Poll the API until the job is completed, then return ``True`` if the + job is succesful, else ``False``. + + Args: + job_id (int or str): The id of the job + d (int): The time delta between successive polls in seconds. + Default 1. + + Returns: + success (bool): True if the job status is success, else False. + + Raises: + requests.HTTPError: If any API request fails. + """ + completed_statuses = ('error', 'success') + while True: + status = self.status(job_id) + if status in completed_statuses: + return status == 'success' diff --git a/kbcstorage/workspaces.py b/kbcstorage/workspaces.py new file mode 100644 index 0000000..802334c --- /dev/null +++ b/kbcstorage/workspaces.py @@ -0,0 +1,164 @@ +""" +Manages calls to the Storage API relating to workspaces. + +Full documentation `here`. + +.. _here: + http://docs.keboola.apiary.io/#reference/workspaces/ +""" +from kbcstorage.base import Endpoint + + +def _make_body(mapping): + """ + Given a dict mapping Keboola tables to aliases, construct the body of + the HTTP request to load said tables. + + Args: + mapping(:obj:`dict`): Keys contain the full names of the tables to + be loaded (ie. 'in.c-bucker.table_name') and values contain the + aliases to which they will be loaded (ie. 'table_name'). + """ + body = {} + template = 'input[{0}][{1}]' + for i, (k, v) in enumerate(mapping.items()): + body[template.format(i, 'source')] = k + body[template.format(i, 'destination')] = v + + return body + + +class Workspaces(Endpoint): + """ + Workspaces Endpoint + """ + def __init__(self, root_url, token): + """ + Create a Workspaces endpoint. + + Args: + url (:obj:`str`): The base url for the API. + token (:obj:`str`): A storage API key. + """ + super().__init__(root_url, 'workspaces', token) + + def list(self): + """ + List the details of all workspaces in the project. + + Returns: + response_body: The json from the HTTP response. + + Raises: + requests.HTTPError: If the API request fails. + """ + headers = {'X-StorageApi-Token': self.token} + return self.get(self.base_url, headers=headers) + + def detail(self, workspace_id): + """ + Retrieves information about a given workspace. + + Note that the password to the workspace can only be retrieved when the + workspace is created. + + Args: + workspace_id (int or str): The id of the workspace. + + Raises: + requests.HTTPError: If the API request fails. + """ + headers = {'X-StorageApi-Token': self.token} + url = '{}/{}'.format(self.base_url, workspace_id) + return self.get(url, headers=headers) + + def create(self, backend=None, timeout=None): + """ + Create a new Workspace and return the credentials. + + Args: + backend (:obj:`str`): The type of engine for the workspace. + 'redshift' or 'snowflake'. Default redshift. + timeout (int): The timeout, in seconds, for SQL statements. + Only supported by snowflake backends. + + Raises: + requests.HTTPError: If the API request fails. + """ + headers = { + 'X-StorageApi-Token': self.token, + 'Content-Type': 'application/x-www-form-urlencoded' + } + body = { + 'backend': backend, + 'statementTimeoutSeconds': timeout + } + + return self.post(self.base_url, data=body, headers=headers) + + def delete(self, workspace_id): + """ + Deletes a workspace. + + This also irreversibly removes workspace content. + + Args: + workspace_id (int or str): The id of the workspace to be deleted. + + Raises: + requests.HTTPError: If the API request fails. + """ + headers = { + 'X-StorageApi-Token': self.token, + 'Content-Type': 'application/x-www-form-urlencoded' + } + url = '{}/{}'.format(self.base_url, workspace_id) + + # This shadows the superclass... + return super().delete(url, headers=headers) + + def reset_password(self, workspace_id): + """ + Generate a new password for the workspace. + + Args: + workspace_id (int or str): The id of the workspace for which the + password should be reset. + + Raises: + requests.HTTPError: If the API request fails. + """ + headers = { + 'X-StorageApi-Token': self.token, + 'Content-Type': 'application/x-www-form-urlencoded' + } + url = '{}/{}/password'.format(self.base_url, workspace_id) + return self.post(url, headers=headers) + + def load_tables(self, workspace_id, table_mapping, preserve=None): + """ + Load tabes from storage into a workspace. + + Args: + workspace_id (int or str): The id of the workspace to which to load + the tables. + table_mapping (:obj:`dict`): Source table names mapped to + destination table names. + preserve (bool): If False, drop tables, else keep tables in + workspace. + + Raises: + requests.HTTPError: If the API request fails. + + Todo: + * Column data types. + """ + headers = { + 'X-StorageApi-Token': self.token, + 'Content-Type': 'application/x-www-form-urlencoded' + } + body = _make_body(table_mapping) + body['preserve'] = preserve + url = '{}/{}/load'.format(self.base_url, workspace_id) + + return self.post(url, data=body, headers=headers) diff --git a/setup.py b/setup.py index dfe1084..c8ebcdf 100644 --- a/setup.py +++ b/setup.py @@ -9,11 +9,13 @@ setup_requires=['setuptools_scm'], url='https://github.com/keboola/sapi-python-client', download_url='https://github.com/keboola/sapi-python-client', - packages=find_packages(), + packages=find_packages(exclude=['tests']), install_requires=[ 'boto3', 'requests' ], + test_suite='tests', + tests_require=['responses'], long_description=readme, license="MIT" ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/bucket_responses.py b/tests/bucket_responses.py new file mode 100644 index 0000000..69a6768 --- /dev/null +++ b/tests/bucket_responses.py @@ -0,0 +1,63 @@ +list_response = [ + { + "uri": "https://connection.keboola.com/v2/storage/buckets/in.c-main", + "id": "in.c-main", + "name": "c-main", + "stage": "in", + "description": "Main user storage", + "tables": ("https://connection.keboola.com/v2/storage/buckets/" + "in.c-main/tables"), + "backend": "snowflake" + }, + { + "uri": ("https://connection.keboola.com/v2/storage/buckets/" + "in.c-organizationData"), + "id": "in.c-organizationData", + "name": "c-organizationData", + "stage": "in", + "description": "Source bucket description", + "tables": ("https://connection.keboola.com/v2/storage/buckets/" + "in.c-organizationData/tables"), + "backend": "snowflake", + "isReadonly": True, + "sourceBucket": { + "id": "in.c-main", + "name": "c-main", + "description": "Organization shared data", + "project": { + "id": 123, + "name": "Project name" + } + } + } +] + +detail_response = { + "uri": "https://connection.keboola.com/v2/storage/buckets/in.c-ga", + "id": "in.c-ga", + "name": "c-ga", + "stage": "in", + "description": "Google Analytics", + "tables": "https://connection.keboola.com/v2/storage/buckets/in.c-ga/tables", + "backend": "mysql" +} + +create_response = { + "uri": ("https://connection.keboola.com/v2/storage/buckets/" + "in.c-my-new-bucket"), + "id": "in.c-my-new-bucket", + "name": "c-my-new-bucket", + "stage": "in", + "description": "Some Description", + "tables": ("https://connection.keboola.com/v2/storage/buckets/" + "in.c-my-new-bucket/tables"), + "created": "2017-02-13T12:01:05+0100", + "lastChangeDate": None, + "isReadOnly": False, + "dataSizeBytes": 0, + "rowsCount": 0, + "isMaintenance": False, + "backend": "snowflake", + "sharing": None, + "attributes": [] +} diff --git a/tests/job_responses.py b/tests/job_responses.py new file mode 100644 index 0000000..a702d82 --- /dev/null +++ b/tests/job_responses.py @@ -0,0 +1,73 @@ +list_response = [ + { + "id": 22077337, + "status": "success", + "url": "https://connection.keboola.com/v2/storage/jobs/22077337", + "tableId": None, + "operationName": "workspaceLoad", + "operationParams": { + "workspaceId": "78423", + "preserve": False, + "input": [ + { + "source": "in.c-application-testing.cashier-data", + "destination": "my-table" + } + ], + "queue": "main_fast" + }, + "createdTime": "2017-02-13T16:41:18+0100", + "startTime": "2017-02-13T16:41:18+0100", + "endTime": "2017-02-13T16:42:00+0100", + "runId": None, + "results": None, + "creatorToken": { + "id": "27978", + "description": "ondrej.popelka@keboola.com" + }, + "metrics": { + "inCompressed": False, + "inBytes": 0, + "inBytesUncompressed": 0, + "outCompressed": True, + "outBytes": 7168, + "outBytesUncompressed": 0 + } + } +] + +detail_response = { + "id": 22077337, + "status": "success", + "url": "https://connection.keboola.com/v2/storage/jobs/22077337", + "tableId": None, + "operationName": "workspaceLoad", + "operationParams": { + "workspaceId": "78423", + "preserve": False, + "input": [ + { + "source": "in.c-application-testing.cashier-data", + "destination": "my-table" + } + ], + "queue": "main_fast" + }, + "createdTime": "2017-02-13T16:41:18+0100", + "startTime": "2017-02-13T16:41:18+0100", + "endTime": "2017-02-13T16:42:00+0100", + "runId": None, + "results": None, + "creatorToken": { + "id": "27978", + "description": "ondrej.popelka@keboola.com" + }, + "metrics": { + "inCompressed": False, + "inBytes": 0, + "inBytesUncompressed": 0, + "outCompressed": True, + "outBytes": 7168, + "outBytesUncompressed": 0 + } +} diff --git a/tests/test_base.py b/tests/test_base.py new file mode 100644 index 0000000..5d60dea --- /dev/null +++ b/tests/test_base.py @@ -0,0 +1,62 @@ +import unittest + +from requests import HTTPError + +from kbcstorage.base import Endpoint + + +class TestEndpoint(unittest.TestCase): + """ + Test Endpoint functionality. + """ + def setUp(self): + self.root = 'https://httpbin.org' + self.token = '' + + def test_get(self): + """ + Simple get works. + """ + endpoint = Endpoint(self.root, 'get', self.token) + requested_url = endpoint.get(endpoint.base_url)['url'] + assert requested_url == 'https://httpbin.org/get' + + def test_get_404(self): + """ + Get inexistent resource raises HTTPError. + """ + endpoint = Endpoint(self.root, 'get', self.token) + with self.assertRaises(HTTPError): + endpoint.get('{}/not-a-url'.format(endpoint.base_url)) + + def test_post(self): + """ + Simple post works. + """ + endpoint = Endpoint(self.root, 'post', self.token) + requested_url = endpoint.post(endpoint.base_url)['url'] + assert requested_url == 'https://httpbin.org/post' + + def test_post_404(self): + """ + Post to inexistent resource raises HTTPError. + """ + endpoint = Endpoint(self.root, 'post', self.token) + with self.assertRaises(HTTPError): + endpoint.post('{}/not-a-url'.format(endpoint.base_url)) + + def test_delete(self): + """ + Simple delete works. + """ + endpoint = Endpoint(self.root, 'delete', self.token) + resp = endpoint.delete(endpoint.base_url) + assert resp is None + + def test_delete_404(self): + """ + Delete inexistent resource raises HTTPError. + """ + endpoint = Endpoint(self.root, 'delete', self.token) + with self.assertRaises(HTTPError): + endpoint.delete('{}/not-a-url'.format(endpoint.base_url)) diff --git a/tests/test_buckets.py b/tests/test_buckets.py index 456e694..809d871 100644 --- a/tests/test_buckets.py +++ b/tests/test_buckets.py @@ -1,30 +1,118 @@ -from kbcstorage.client import Client import os import unittest + from requests import exceptions +import responses + +from kbcstorage.buckets import Buckets +from kbcstorage.client import Client + +from .bucket_responses import list_response, detail_response, create_response class TestBuckets(unittest.TestCase): def setUp(self): - self.client = Client(os.getenv('KBC_TEST_TOKEN'), os.getenv('KBC_TEST_API_URL')) + self.client = Client(os.getenv('KBC_TEST_API_URL'), + os.getenv('KBC_TEST_TOKEN')) try: - self.client.drop_bucket('in.c-py-test', {'force': True}) + self.client.buckets.delete('in.c-py-test', force=True) except exceptions.HTTPError as e: if e.response.status_code != 404: raise def tearDown(self): try: - self.client.drop_bucket('in.c-py-test', {'force': True}) + self.client.drop_bucket('in.c-py-test', force=True) except exceptions.HTTPError as e: if e.response.status_code != 404: raise def test_create_bucket(self): - bucket_id = self.client.create_bucket('py-test', 'in', 'Test bucket') - self.assertTrue(self.client.bucket_exists(bucket_id)) + bucket_id = self.client.buckets.create(name='py-test', + stage='in', + description='Test bucket')['id'] + self.assertTrue(self.client.buckets.exists(bucket_id)) def test_bucket_exists(self): - bucket_id = self.client.create_bucket('py-test', 'in', 'Test bucket') - self.assertTrue(self.client.bucket_exists(bucket_id)) - self.assertFalse(self.client.bucket_exists('some-totally-non-existent-bucket')) + bucket_id = self.client.buckets.create(name='py-test', + stage='in', + description='Test bucket')['id'] + self.assertTrue(self.client.buckets.exists(bucket_id)) + self.assertFalse( + self.client.buckets.exists('some-totally-non-existent-bucket') + ) + + +class TestBucketsWithMocks(unittest.TestCase): + def setUp(self): + token = 'dummy_token' + base_url = 'https://connection.keboola.com/v2/storage/' + self.buckets = Buckets(base_url, token) + + @responses.activate + def test_list(self): + """ + Buckets mocks list correctly. + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/buckets', + json=list_response + ) + ) + buckets_list = self.buckets.list() + assert isinstance(buckets_list, list) + + @responses.activate + def test_detail_by_id(self): + """ + Buckets mocks detail by integer id correctly. + """ + responses.add( + responses.Response( + method='GET', + url=('https://connection.keboola.com/v2/storage/buckets/' + 'in.c-ga'), + json=detail_response + ) + ) + bucket_id = 'in.c-ga' + bucket_detail = self.buckets.detail(bucket_id) + assert bucket_detail['id'] == 'in.c-ga' + + @responses.activate + def test_delete(self): + """ + Buckets mock deletes bucket by id. + """ + responses.add( + responses.Response( + method='DELETE', + url='https://connection.keboola.com/v2/storage/buckets/1', + json={} + ) + ) + bucket_id = '1' + deleted_detail = self.buckets.delete(bucket_id) + assert deleted_detail is None + + @responses.activate + def test_create(self): + """ + Buckets mock creates new bucket. + """ + responses.add( + responses.Response( + method='POST', + url='https://connection.keboola.com/v2/storage/buckets', + json=create_response + ) + ) + name = 'my-new-bucket' + description = 'Some Description' + backend = 'snowflake' + created_detail = self.buckets.create(name=name, + description=description, + backend=backend) + assert created_detail['id'] == 'in.c-{}'.format(name) diff --git a/tests/test_jobs.py b/tests/test_jobs.py new file mode 100644 index 0000000..1195f66 --- /dev/null +++ b/tests/test_jobs.py @@ -0,0 +1,158 @@ +""" +Test basic functionality of the Jobs endpoint +""" +import unittest + +import responses + +from kbcstorage.jobs import Jobs + +from .job_responses import list_response, detail_response + + +class TestJobsEndpointWithMocks(unittest.TestCase): + """ + Test the methods of a Jobs endpoint instance with mock HTTP responses + """ + def setUp(self): + token = 'dummy_token' + base_url = 'https://connection.keboola.com/v2/storage/' + self.jobs = Jobs(base_url, token) + + @responses.activate + def test_list(self): + """ + Jobs mocks list correctly + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/jobs', + json=list_response + ) + ) + jobs_list = self.jobs.list() + assert isinstance(jobs_list, list) + + @responses.activate + def test_detail_by_id(self): + """ + Jobs Endpoint can mock detail by integer id + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/jobs/22077337', + json=detail_response + ) + ) + job_id = 22077337 + job_detail = self.jobs.detail(job_id) + assert job_detail['id'] == 22077337 + + @responses.activate + def test_job_status(self): + """ + Jobs mock status works correctly. + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/jobs/22077337', + json=detail_response + ) + ) + job_id = 22077337 + job_status = self.jobs.status(job_id) + assert job_status == 'success' + + @responses.activate + def test_job_completion(self): + """ + Jobs mock completion check works correctly. + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/jobs/22077337', + json=detail_response + ) + ) + job_id = 22077337 + job_completed = self.jobs.completed(job_id) + assert job_completed is True + + @responses.activate + def test_job_blocking(self): + """ + Jobs mock blocking polls until completion. + """ + for _ in range(5): + responses.add( + responses.Response( + method='GET', + url=('https://connection.keboola.com/v2/storage/jobs/' + '22077337'), + json={'status': 'processing'} + ) + ) + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/jobs/22077337', + json=detail_response + ) + ) + job_id = 22077337 + self.jobs.block_until_completed(job_id, d=0.000001) + assert True + + @responses.activate + def test_success_blocking_if_success(self): + """ + Jobs mock blocking polls until completion. + """ + for _ in range(5): + responses.add( + responses.Response( + method='GET', + url=('https://connection.keboola.com/v2/storage/jobs/' + '22077337'), + json={'status': 'processing'} + ) + ) + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/jobs/22077337', + json={'status': 'success'} + ) + ) + job_id = 22077337 + success = self.jobs.block_for_success(job_id, d=0.000001) + assert success is True + + @responses.activate + def test_success_blocking_if_error(self): + """ + Jobs mock blocking polls until completion. + """ + for _ in range(5): + responses.add( + responses.Response( + method='GET', + url=('https://connection.keboola.com/v2/storage/jobs/' + '22077337'), + json={'status': 'processing'} + ) + ) + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/jobs/22077337', + json={'status': 'error'} + ) + ) + job_id = 22077337 + success = self.jobs.block_for_success(job_id, d=0.000001) + assert success is False diff --git a/tests/test_workspaces.py b/tests/test_workspaces.py new file mode 100644 index 0000000..4ecd273 --- /dev/null +++ b/tests/test_workspaces.py @@ -0,0 +1,219 @@ +""" +Asses basic functionality of the Workspace endpoint. +""" +import unittest + +import responses +from requests import HTTPError + +from kbcstorage.workspaces import Workspaces + +from .workspace_responses import (list_response, detail_response, + load_tables_response, create_response, + reset_password_response) + + +class TestWorkspacesEndpointWithMocks(unittest.TestCase): + """ + Test the methods of a Workspaces endpoint instance with mock HTTP responses + """ + def setUp(self): + token = 'dummy_token' + base_url = 'https://connection.keboola.com/v2/storage/' + self.ws = Workspaces(base_url, token) + + @responses.activate + def test_list(self): + """ + Workspace mocks list correctly + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/workspaces', + json=list_response + ) + ) + workspace_list = self.ws.list() + assert isinstance(workspace_list, list) + + @responses.activate + def test_detail_by_integer_id(self): + """ + Workspace Endpoint can mock detail by integer id + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/workspaces/1', + json=detail_response + ) + ) + workspace_id = 1 + workspace_detail = self.ws.detail(workspace_id) + assert workspace_detail['id'] == 1 + + @responses.activate + def test_detail_by_str_id(self): + """ + Workspace Endpoint can get mocked detail by string id + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/workspaces/1', + json=detail_response + ) + ) + workspace_id = '1' + workspace_detail = self.ws.detail(workspace_id) + assert workspace_detail['id'] == 1 + + @responses.activate + def test_detail_inexsitent_workspace(self): + """ + Workspace Endpoint raises HTTPError when mocking inexistent workspace + detail + """ + msg = ('404 Client Error: Not Found for url: ' + 'https://connection.keboola.com/v2/storage/workspaces/1') + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/workspaces/1', + body=HTTPError(msg) + ) + ) + workspace_id = '1' + with self.assertRaises(HTTPError) as error_context: + self.ws.detail(workspace_id) + assert error_context.exception.args[0] == msg + + @responses.activate + def test_create(self): + """ + Workspace endpoint mock creates new workspace + """ + responses.add( + responses.Response( + method='POST', + url='https://connection.keboola.com/v2/storage/workspaces', + json=create_response + ) + ) + created_detail = self.ws.create() + assert created_detail['connection']['password'] == 'abc' + + @responses.activate + def test_delete(self): + """ + Workspace endpoint mock deletes workspace by id + """ + responses.add( + responses.Response( + method='DELETE', + url='https://connection.keboola.com/v2/storage/workspaces/1', + json=create_response + ) + ) + workspace_id = '1' + deleted_detail = self.ws.delete(workspace_id) + assert deleted_detail is None + + @responses.activate + def test_delete_inexistent_workspace_raises_404(self): + """ + Workspace endpoint raises 404 when mock deleting inexsitent workspace + """ + msg = ('404 Client Error: Not Found for url: ' + 'https://connection.keboola.com/v2/storage/workspaces/1') + responses.add( + responses.Response( + method='DELETE', + url='https://connection.keboola.com/v2/storage/workspaces/1', + body=HTTPError(msg) + ) + ) + workspace_id = '1' + with self.assertRaises(HTTPError) as error_context: + self.ws.delete(workspace_id) + assert error_context.exception.args[0] == msg + + @responses.activate + def test_load_tables_to_workspace(self): + """ + Workspace endpoint mock loads table + """ + responses.add( + responses.Response( + method='POST', + url=('https://connection.keboola.com/v2/storage/workspaces/' + '78432/load'), + json=load_tables_response + ) + ) + workspace_id = '78432' + mapping = {"in.c-application-testing.cashier-data": "my-table"} + loaded_detail = self.ws.load_tables(workspace_id, mapping) + assert loaded_detail['id'] == 22077337 + + @responses.activate + def test_load_inexistent_tables_to_workspace(self): + """ + Workspace endpoint raises HTTPError when mock loading inexistent table + """ + msg = ('404 Client Error: Not Found for url: ' + 'https://connection.keboola.com/v2/storage/workspaces/78432/' + 'load') + responses.add( + responses.Response( + method='POST', + url=('https://connection.keboola.com/v2/storage/workspaces/' + '78432/load'), + body=HTTPError(msg) + ) + ) + workspace_id = '78432' + mapping = {"in.c-table.does_not_exist": "my-table"} + with self.assertRaises(HTTPError) as error_context: + self.ws.load_tables(workspace_id, mapping) + assert error_context.exception.args[0] == msg + + @responses.activate + def test_reset_workspace_password(self): + """ + Workspace endpoint mock resets password for workspace + """ + responses.add( + responses.Response( + method='POST', + url=('https://connection.keboola.com/v2/storage/workspaces/' + '1/password'), + json=reset_password_response + ) + ) + workspace_id = '1' + reset_detail = self.ws.reset_password(workspace_id) + assert reset_detail['password'] == 'top_secret_password' + + @responses.activate + def test_reset_password_for_inexistent_workspace(self): + """ + Workspace endpoint raises HTTPError when mock resetting password for + inexistent workspace + """ + msg = ('404 Client Error: Not Found for url: ' + 'https://connection.keboola.com/v2/storage/workspaces/1/' + 'password') + responses.add( + responses.Response( + method='POST', + url=('https://connection.keboola.com/v2/storage/workspaces/' + '1/password'), + body=HTTPError(msg) + ) + ) + workspace_id = '1' + with self.assertRaises(HTTPError) as error_context: + self.ws.reset_password(workspace_id) + assert error_context.exception.args[0] == msg diff --git a/tests/workspace_responses.py b/tests/workspace_responses.py new file mode 100644 index 0000000..f2742db --- /dev/null +++ b/tests/workspace_responses.py @@ -0,0 +1,114 @@ +list_response = [ + { + "id": 234, + "name": "boring_wozniak", + "component": "wr-db", + "configurationId": "aws-1", + "created": "2016-05-17T11:11:20+0200", + "connection": { + "backend": "snowflake", + "host": "keboola.snowflakecomputing.com", + "database": "keboola_123", + "schema": "boring_wozniak", + "warehouse": "SAPI_PROD", + "user": "xzy" + }, + "creatorToken": { + "id": 234, + "description": "martin@keboola.com" + }, + "creatorUser": { + "id": 234, + "name": "Martin" + } + } +] + +detail_response = { + "id": 1, + "name": "boring_wozniak", + "component": "wr-db", + "configurationId": "aws-1", + "created": "2016-05-17T11:11:20+0200", + "connection": { + "backend": "snowflake", + "host": "keboola.snowflakecomputing.com", + "database": "keboola_123", + "schema": "boring_wozniak", + "warehouse": "SAPI_PROD", + "user": "xzy" + }, + "creatorToken": { + "id": 234, + "description": "martin@keboola.com" + }, + "creatorUser": { + "id": 234, + "name": "Martin" + } +} + +create_response = { + "id": 234, + "name": "boring_wozniak", + "component": "wr-db", + "configurationId": "aws-1", + "created": "2016-05-17T11:11:20+0200", + "connection": { + "backend": "snowflake", + "host": "keboola.snowflakecomputing.com", + "database": "keboola_123", + "schema": "boring_wozniak", + "warehouse": "SAPI_PROD", + "user": "xzy", + "password": "abc" + }, + "creatorToken": { + "id": 234, + "description": "martin@keboola.com" + }, + "creatorUser": { + "id": 234, + "name": "Martin" + } +} + +load_tables_response = { + "id": 22077337, + "status": "waiting", + "url": "https://connection.keboola.com/v2/storage/jobs/22077337", + "tableId": None, + "operationName": "workspaceLoad", + "operationParams": { + "workspaceId": "78423", + "preserve": False, + "input": [ + { + "source": "in.c-application-testing.cashier-data", + "destination": "my-table" + } + ], + "queue": "main_fast" + }, + "createdTime": "2017-02-13T16:41:18+0100", + "startTime": None, + "endTime": None, + "runId": None, + "results": None, + "creatorToken": { + "id": "27978", + "description": "ondrej.popelka@keboola.com" + }, + "metrics": { + "inCompressed": False, + "inBytes": 0, + "inBytesUncompressed": 0, + "outCompressed": False, + "outBytes": 0, + "outBytesUncompressed": 0 + } +} + +reset_password_response = { + "password": "top_secret_password", +}