diff --git a/.github/workflows/push_request.yml b/.github/workflows/push_request.yml new file mode 100644 index 00000000..b4de41c6 --- /dev/null +++ b/.github/workflows/push_request.yml @@ -0,0 +1,72 @@ +name: On Push + +# If this is made manual, then also update the pull_request.yml, to run builds +# on main on merge. +on: + push: + branches: [ main ] + + # Allows you to run this workflow manually from the Actions tab + workflow_dispatch: + +jobs: + release: + runs-on: ubuntu-latest + outputs: + version: ${{ steps.release.outputs.release }} + steps: + - uses: actions/checkout@v2 + + - name: Set release + id: semrel + uses: go-semantic-release/action@v1 + with: + github-token: ${{ secrets.GITHUB_TOKEN }} + allow-initial-development-versions: true + force-bump-patch-version: true + + - name: Output release + id: release + run: echo "::set-output name=release::${{ steps.semrel.outputs.version }}" + + publish_python: + # The type of runner that the job will run on + runs-on: ubuntu-latest + needs: release + + # Steps represent a sequence of tasks that will be executed as part of the job + steps: + - uses: actions/checkout@v2 + + - name: Setup Python + uses: actions/setup-python@v2 + with: + python-version: '3.x' + + - name: Install python dependencies + run: | + python -m pip install --upgrade pip + pip install setuptools wheel twine build + - name: Configure Artifactory publish credentials + run: | + set -e + set -x + pushd artifactory_setup + ./jfrog_setup.sh deploy-user ${{ secrets.PUBLIC_ARTIFACTORY_DEPLOY_USER_MAGIC_PASSTOKEN }} >> ~/.pypirc + popd + shell: bash + + - name: Build and publish package + run: | + set -e + set -x + export MOMENTO_SDK_VERSION="${{needs.release.outputs.version}}" + if [ -z "$MOMENTO_SDK_VERSION"] + then + echo "Using default version" + export MOMENTO_SDK_VERSION="0.0.dev"`date +%s` + fi + echo "MOMENTO_SDK_VERSION=${MOMENTO_SDK_VERSION}" + python -m build + twine upload -r local dist/* --config-file ~/.pypirc + shell: bash diff --git a/.gitignore b/.gitignore index d93d06ff..3829d555 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,7 @@ # Compiled python modules. src/*.pyc +src/momento/*.pyc +src/momento/__pycache__ # Setuptools distribution folder. /dist/ diff --git a/README.md b/README.md index 1e3b9545..eb059c11 100644 --- a/README.md +++ b/README.md @@ -12,4 +12,20 @@ The script creates a python virtual environment and installs dependencies ## Setting up IDE ### Visual Studio Code Use `Cmd` + `Shift` + `P` to search for `Python: Interpreter` and select: -`./client_sdk_python_env/bin/python` \ No newline at end of file +`./client_sdk_python_env/bin/python` + +# Developing +Once your pre-requisites are accomplished + +Run the following command to start your virtual environment + +`source client_sdk_python_env/bin/activate` + +To install the package under development + +`pip install -e .` + +To test your changes you can then just run your python shell as follows: + +`python` this will start the interactive shell or if you prefer you may put all +your code in a my_test.py file and run `python my_test.py` diff --git a/artifactory_setup/jfrog_setup.sh b/artifactory_setup/jfrog_setup.sh new file mode 100755 index 00000000..e7b8a499 --- /dev/null +++ b/artifactory_setup/jfrog_setup.sh @@ -0,0 +1,11 @@ +#!/bin/bash +# Configure Git Hub Workflow environment for Uploading to JFrog +set -e +set -x +echo "[distutils]" +echo "index-servers = local" +echo "[local]" +echo "repository: https://momento.jfrog.io/artifactory/api/pypi/pypi-local" +echo "username: $1" +echo "password: $2" +echo "" diff --git a/setup.cfg b/setup.cfg index b0d073bd..e8220b5b 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,5 +1,5 @@ [metadata] -name = momento-sdk +name = momento [options] package_dir = @@ -7,9 +7,11 @@ package_dir = packages = find: python_requires = >=3.6 install_requires = - momento-wire-types==0.3.0 + momento-wire-types==0.6.0 build setuptools + pyjwt + grpcio [options.packages.find] where = src diff --git a/setup.py b/setup.py index 2a836a9c..6a8bdb14 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ import os import time -version = os.getenv("PYPI_MOMENTO_WIRE_TYPE_VERSION") +version = os.getenv("MOMENTO_SDK_VERSION") if [version == None]: version = '0.0.dev' diff --git a/src/momento_sdk/__init__.py b/src/momento/__init__.py similarity index 100% rename from src/momento_sdk/__init__.py rename to src/momento/__init__.py diff --git a/src/momento/_authorization_interceptor.py b/src/momento/_authorization_interceptor.py new file mode 100644 index 00000000..c32f7df8 --- /dev/null +++ b/src/momento/_authorization_interceptor.py @@ -0,0 +1,6 @@ +from . import _header_client_interceptor + + +def get_authorization_interceptor(auth_token): + return _header_client_interceptor.header_adder_interceptor( + 'authorization', auth_token) diff --git a/src/momento/_cache_name_interceptor.py b/src/momento/_cache_name_interceptor.py new file mode 100644 index 00000000..1979f922 --- /dev/null +++ b/src/momento/_cache_name_interceptor.py @@ -0,0 +1,6 @@ +from . import _header_client_interceptor + + +def get_cache_name_interceptor(cache_name): + return _header_client_interceptor.header_adder_interceptor( + 'cache', cache_name) diff --git a/src/momento/_cache_service_errors_converter.py b/src/momento/_cache_service_errors_converter.py new file mode 100644 index 00000000..0b0d80a4 --- /dev/null +++ b/src/momento/_cache_service_errors_converter.py @@ -0,0 +1,40 @@ +import grpc +from . import errors +from momento_wire_types import cacheclient_pb2 as cache_client_types + +__rpc_to_error = { + grpc.StatusCode.ALREADY_EXISTS: errors.CacheExistsError, + grpc.StatusCode.INVALID_ARGUMENT: errors.CacheValueError, + grpc.StatusCode.NOT_FOUND: errors.CacheNotFoundError, + grpc.StatusCode.PERMISSION_DENIED: errors.PermissionError, +} + +# Till the time MR2 stops returning errors in Enums +__ecache_result_to_error = { + cache_client_types.Bad_Request: errors.CacheValueError, + cache_client_types.Internal_Server_Error: errors.InternalServerError, + cache_client_types.Service_Unavailable: errors.InternalServerError, + cache_client_types.Unauthorized: errors.PermissionError, +} + + +def convert(exception): + if (isinstance(exception, errors.SdkError)): + return exception + + if (isinstance(exception, grpc.RpcError)): + if exception.code() in __rpc_to_error: + return __rpc_to_error[exception.code()](exception.details()) + else: + return errors.InternalServerError( + 'CacheService failed with an internal error') + + return errors.ClientSdkError('Operation failed with error: ' + + str(exception)) + + +def convert_ecache_result(ecache_result, message): + if (ecache_result in __ecache_result_to_error): + return __ecache_result_to_error[ecache_result](message) + return errors.InternalServerError( + 'CacheService failed with an internal error') diff --git a/src/momento/_generic_client_interceptor.py b/src/momento/_generic_client_interceptor.py new file mode 100644 index 00000000..fd5be7fe --- /dev/null +++ b/src/momento/_generic_client_interceptor.py @@ -0,0 +1,17 @@ +import grpc + + +class _GenericClientInterceptor(grpc.UnaryUnaryClientInterceptor): + def __init__(self, interceptor_function): + self._fn = interceptor_function + + def intercept_unary_unary(self, continuation, client_call_details, + request): + new_details, new_request_iterator, postprocess = self._fn( + client_call_details, iter((request, )), False, False) + response = continuation(new_details, next(new_request_iterator)) + return postprocess(response) if postprocess else response + + +def create(intercept_call): + return _GenericClientInterceptor(intercept_call) diff --git a/src/momento/_header_client_interceptor.py b/src/momento/_header_client_interceptor.py new file mode 100644 index 00000000..6c776367 --- /dev/null +++ b/src/momento/_header_client_interceptor.py @@ -0,0 +1,30 @@ +import collections +import grpc + +from . import _generic_client_interceptor + + +class _ClientCallDetails( + collections.namedtuple( + '_ClientCallDetails', + ('method', 'timeout', 'metadata', 'credentials')), + grpc.ClientCallDetails): + pass + + +def header_adder_interceptor(header, value): + def intercept_call(client_call_details, request_iterator, + request_streaming, response_streaming): + metadata = [] + if client_call_details.metadata is not None: + metadata = list(client_call_details.metadata) + metadata.append(( + header, + value, + )) + client_call_details = _ClientCallDetails( + client_call_details.method, client_call_details.timeout, metadata, + client_call_details.credentials) + return client_call_details, request_iterator, None + + return _generic_client_interceptor.create(intercept_call) diff --git a/src/momento/_momento_endpoint_resolver.py b/src/momento/_momento_endpoint_resolver.py new file mode 100644 index 00000000..81cb4e73 --- /dev/null +++ b/src/momento/_momento_endpoint_resolver.py @@ -0,0 +1,31 @@ +import jwt +from jwt.exceptions import DecodeError + +from . import errors + +_MOMENTO_CONTROL_ENDPOINT_PREFIX = 'control.' +_MOMENTO_CACHE_ENDPOINT_PREFIX = 'cache.' +_CONTROL_ENDPOINT_CLAIM_ID = 'cp' +_CACHE_ENDPOINT_CLAIM_ID = 'c' + + +class _Endpoints: + def __init__(self, control_endpoint, cache_endpoint): + self.control_endpoint = control_endpoint + self.cache_endpoint = cache_endpoint + + +def resolve(auth_token, endpoint_override=None): + if (endpoint_override is not None): + return _Endpoints(_MOMENTO_CONTROL_ENDPOINT_PREFIX + endpoint_override, + _MOMENTO_CACHE_ENDPOINT_PREFIX + endpoint_override) + return _getEndpointFromToken(auth_token) + + +def _getEndpointFromToken(auth_token): + try: + claims = jwt.decode(auth_token, options={'verify_signature': False}) + return _Endpoints(claims[_CONTROL_ENDPOINT_CLAIM_ID], + claims[_CACHE_ENDPOINT_CLAIM_ID]) + except (DecodeError, KeyError): + raise errors.InvalidInputError('Invalid Auth token.') from None diff --git a/src/momento/cache.py b/src/momento/cache.py new file mode 100644 index 00000000..36451af7 --- /dev/null +++ b/src/momento/cache.py @@ -0,0 +1,89 @@ +import grpc +import time +import uuid +import momento_wire_types.cacheclient_pb2_grpc as cache_client +import momento_wire_types.cacheclient_pb2 as cache_client_types + +from . import _cache_service_errors_converter +from . import _authorization_interceptor +from . import _cache_name_interceptor +from . import errors +from . import cache_operation_responses as cache_sdk_resp + + +class Cache: + def __init__(self, auth_token, cache_name, endpoint, default_ttlSeconds): + self._validate_ttl(default_ttlSeconds) + self._default_ttlSeconds = default_ttlSeconds + self._secure_channel = grpc.secure_channel( + endpoint, grpc.ssl_channel_credentials()) + auth_interceptor = _authorization_interceptor.get_authorization_interceptor( + auth_token) + cache_interceptor = _cache_name_interceptor.get_cache_name_interceptor( + cache_name) + intercept_channel = grpc.intercept_channel(self._secure_channel, + auth_interceptor, + cache_interceptor) + self._client = cache_client.ScsStub(intercept_channel) + self._wait_until_ready() + + # Temporary measure + def _wait_until_ready(self): + start_time = time.time() + max_wait_seconds = 5 + back_off_millis = 50 + last_exception = None + + while (time.time() - start_time < max_wait_seconds): + try: + self.get(uuid.uuid1().bytes) + return + except Exception as e: + last_exception = e + time.sleep(back_off_millis / 1000.0) + + raise _cache_service_errors_converter.convert(last_exception) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, exc_traceback): + self._secure_channel.close() + + def set(self, key, value, ttl_seconds=None): + try: + item_ttl_seconds = self._default_ttlSeconds if ttl_seconds is None else ttl_seconds + self._validate_ttl(item_ttl_seconds) + set_request = cache_client_types.SetRequest() + set_request.cache_key = self._asBytes( + key, 'Unsupported type for key: ') + set_request.cache_body = self._asBytes( + value, 'Unsupported type for value: ') + set_request.ttl_milliseconds = item_ttl_seconds * 1000 + response = self._client.Set(set_request) + return cache_sdk_resp.CacheSetResponse(response, + set_request.cache_body) + except Exception as e: + raise _cache_service_errors_converter.convert(e) + + def get(self, key): + try: + get_request = cache_client_types.GetRequest() + get_request.cache_key = self._asBytes( + key, 'Unsupported type for key: ') + response = self._client.Get(get_request) + return cache_sdk_resp.CacheGetResponse(response) + except Exception as e: + raise _cache_service_errors_converter.convert(e) + + def _asBytes(self, data, errorMessage): + if (isinstance(data, str)): + return data.encode('utf-8') + if (isinstance(data, bytes)): + return data + raise errors.InvalidInputError(errorMessage + str(type(data))) + + def _validate_ttl(self, ttl_seconds): + if (not isinstance(ttl_seconds, int) or ttl_seconds <= 0): + raise errors.InvalidInputError( + 'TTL Seconds must be a non-zero positive integer.') diff --git a/src/momento/cache_operation_responses.py b/src/momento/cache_operation_responses.py new file mode 100644 index 00000000..35794eb1 --- /dev/null +++ b/src/momento/cache_operation_responses.py @@ -0,0 +1,60 @@ +from enum import Enum +from momento_wire_types import cacheclient_pb2 as cache_client_types +from . import _cache_service_errors_converter as error_converter + + +class CacheResult(Enum): + HIT = 1 + MISS = 2 + + +class CacheSetResponse: + def __init__(self, grpc_set_response, value): + self._value = value + if (grpc_set_response.result is not cache_client_types.Ok): + raise error_converter.convert_ecache_result( + grpc_set_response.result, grpc_set_response.message) + + def str_utf8(self): + return self._value.decode('utf-8') + + def bytes(self): + return self._value + + +class CacheGetResponse: + def __init__(self, grpc_get_response): + self._value = grpc_get_response.cache_body + + if (grpc_get_response.result is not cache_client_types.Hit + and grpc_get_response.result is not cache_client_types.Miss): + raise error_converter.convert_ecache_result( + grpc_get_response.result, grpc_get_response.message) + + if (grpc_get_response.result == cache_client_types.Hit): + self._result = CacheResult.HIT + if (grpc_get_response.result == cache_client_types.Miss): + self._result = CacheResult.MISS + + def str_utf8(self): + if (self._result == CacheResult.HIT): + return self._value.decode('utf-8') + return None + + def bytes(self): + if (self._result == CacheResult.HIT): + return self._value + return None + + def result(self): + return self._result + + +class CreateCacheResponse: + def __init__(self, grpc_create_cache_response): + pass + + +class DeleteCacheResponse: + def __init__(self, grpc_delete_cache_response): + pass diff --git a/src/momento/errors.py b/src/momento/errors.py new file mode 100644 index 00000000..32bab896 --- /dev/null +++ b/src/momento/errors.py @@ -0,0 +1,67 @@ +class SdkError(Exception): + """Base exception for all errors raised by Sdk""" + def __init__(self, message): + super().__init__(message) + + +class ClientSdkError(SdkError): + """For all errors raised by the client. + + Indicates that the request failed on the SDK. The request either did not + make it to the service or if it did the response from the service could + not be parsed successfully. + """ + def __init__(self, message): + super().__init__(message) + + +class InvalidInputError(ClientSdkError): + """Error raised when provided input values to the SDK are invalid + + Some examples - missing required parameters, incorrect parameter + types, malformed input. + """ + def __init__(self, message): + super().__init__(message) + + +class MomentoServiceError(SdkError): + """Errors raised when Momento Service returned an error code""" + def __init__(self, message): + super().__init__(message) + + +class CacheServiceError(MomentoServiceError): + """Base class for all errors raised by the Caching Service""" + def __init__(self, message): + super().__init__(message) + + +class CacheNotFoundError(CacheServiceError): + """Error raised for operations performed on non-existent cache""" + def __init__(self, message): + super().__init__(message) + + +class CacheExistsError(CacheServiceError): + """Error raised when attempting to create a cache with same name""" + def __init__(self, message): + super().__init__(message) + + +class CacheValueError(CacheServiceError): + """Error raised when service validation fails for provided values""" + def __init__(self, message): + super().__init__(message) + + +class PermissionError(CacheServiceError): + """Error when authentication with Cache Service fails""" + def __init__(self, message): + super().__init__(message) + + +class InternalServerError(CacheServiceError): + """Operation failed on the server with an unknown error""" + def __init__(self, message): + super().__init__(message) diff --git a/src/momento/momento.py b/src/momento/momento.py new file mode 100644 index 00000000..c7da12db --- /dev/null +++ b/src/momento/momento.py @@ -0,0 +1,64 @@ +import grpc +import momento_wire_types.controlclient_pb2_grpc as control_client + +from momento_wire_types.controlclient_pb2 import CreateCacheRequest +from momento_wire_types.controlclient_pb2 import DeleteCacheRequest +from . import _cache_service_errors_converter +from . import errors +from .cache import Cache +from . import _authorization_interceptor +from . import _momento_endpoint_resolver +from .cache_operation_responses import CreateCacheResponse +from .cache_operation_responses import DeleteCacheResponse + + +class Momento: + def __init__(self, auth_token, endpoint_override=None): + endpoints = _momento_endpoint_resolver.resolve(auth_token, + endpoint_override) + self._auth_token = auth_token + self._control_endpoint = endpoints.control_endpoint + self._cache_endpoint = endpoints.cache_endpoint + self._secure_channel = grpc.secure_channel( + self._control_endpoint, grpc.ssl_channel_credentials()) + intercept_channel = grpc.intercept_channel( + self._secure_channel, + _authorization_interceptor.get_authorization_interceptor( + auth_token)) + self._client = control_client.ScsControlStub(intercept_channel) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, exc_traceback): + self._secure_channel.close() + + def create_cache(self, cache_name): + try: + request = CreateCacheRequest() + request.cache_name = cache_name + return CreateCacheResponse(self._client.CreateCache(request)) + except Exception as e: + raise _cache_service_errors_converter.convert(e) from None + + def delete_cache(self, cache_name): + try: + request = DeleteCacheRequest() + request.cache_name = cache_name + return DeleteCacheResponse(self._client.DeleteCache(request)) + except Exception as e: + raise _cache_service_errors_converter.convert(e) from None + + def get_cache(self, cache_name, ttl_seconds, create_if_absent=False): + if (create_if_absent): + try: + self.create_cache(cache_name) + except errors.CacheExistsError: + # Cache already exists so nothing to do + pass + return Cache(self._auth_token, cache_name, self._cache_endpoint, + ttl_seconds) + + +def init(auth_token): + return Momento(auth_token) diff --git a/src/momento_sdk/momento.py b/src/momento_sdk/momento.py deleted file mode 100644 index a4b91ba0..00000000 --- a/src/momento_sdk/momento.py +++ /dev/null @@ -1,13 +0,0 @@ -import momento_wire_types.controlclient_pb2 as control_client - -class Momento: - def __init__(self, auth_token, endpoint_override=None): - self._auth_token__ = auth_token - self._endpoint_override = endpoint_override - - def create_cache(self, cache_name): - print("hello") - - -def init(auth_token): - return Momento(auth_token=auth_token)