feat: Base implementation for Cache Operations

.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

.gitignore

@@ -1,5 +1,7 @@
 # Compiled python modules.
 src/*.pyc
+src/momento/*.pyc
+src/momento/__pycache__
 
 # Setuptools distribution folder.
 /dist/

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`
+`./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`

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 ""

setup.cfg

@@ -1,15 +1,17 @@
 [metadata]
-name = momento-sdk
+name = momento
 
 [options]
 package_dir =
     = src
 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

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'

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)

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)

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')

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)

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)

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

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.')