Initial implementation

Author: akajla09

README.md

@@ -1 +1,31 @@
-# warrant-python
+# Warrant Python Library
+
+Use [Warrant](https://warrant.dev/) in Python projects.
+
+## Installation
+
+```python
+pip install warrant-python
+```
+
+## Usage
+
+```python
+import warrant
+
+warrant = Warrant("api_test_f5dsKVeYnVSLHGje44zAygqgqXiLJBICbFzCiAg1E=")
+warrant.create_user()
+```
+
+We’ve used a random API key in these code examples. Replace it with your
+[actual publishable API keys](https://app.warrant.dev) to
+test this code through your own Warrant account.
+
+For more information on how to use the Warrant API, please refer to the
+[Warrant API reference](https://docs.warrant.dev).
+
+Note that we may release new [minor and patch](https://semver.org/) versions of this library with small but backwards-incompatible fixes to the type declarations. These changes will not affect Warrant itself.
+
+## Warrant Documentation
+
+- [Warrant Docs](https://docs.warrant.dev/)

examples/example.py

@@ -0,0 +1,20 @@
+from warrant import Warrant
+
+def make_warrant_requests(api_key):
+    client = Warrant(api_key)
+
+    # Create users and session tokens
+    print("Created user with provided id: " + client.create_user("custom_id_001"))
+    new_user = client.create_user()
+    print("Created user with generated id: " + new_user)
+    print("Created session token: " + client.create_session(new_user))
+
+    # Create and check warrants
+    print(client.create_warrant(object_type="store", object_id="store1", relation="owner", user=new_user))
+    is_authorized = client.is_authorized(object_type="store", object_id="store1", relation="owner", user_to_check=new_user)
+    print(f"New user authorization result: {is_authorized}")
+
+if __name__ == '__main__':
+    # Replace with your Warrant api key
+    api_key = "API_KEY"
+    make_warrant_requests(api_key)

setup.py

@@ -0,0 +1,38 @@
+import pathlib
+from setuptools import setup
+
+DIR = pathlib.Path(__file__).parent
+README = (DIR / "README.md").read_text()
+
+setup(
+    name="warrant-python",
+    version="0.2.0",
+    description="Python SDK for Warrant",
+    long_description=README,
+    long_description_content_type="text/markdown",
+    author="Warrant",
+    author_email="[email protected]",
+    url="https://github.com/warrant-dev/warrant-python",
+    license="Apache Software License (http://www.apache.org/licenses/LICENSE-2.0.txt)",
+    keywords="warrant api authorization access control",
+    packages=["warrant"],
+    install_requires=["requests"],
+    project_urls={
+        "Bug Tracker": "https://github.com/warrant-dev/warrant-python/issues",
+        "Documentation": "https://docs.warrant.dev",
+        "Source Code": "https://github.com/warrant-dev/warrant-python",
+    },
+    classifiers=[
+        "Intended Audience :: Developers",
+        "Operating System :: OS Independent",
+        "Programming Language :: Python",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.4",
+        "Programming Language :: Python :: 3.5",
+        "Programming Language :: Python :: 3.6",
+        "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
+        "Topic :: Software Development :: Libraries :: Python Modules",
+    ],
+)

warrant/__init__.py

@@ -0,0 +1,87 @@
+import requests
+import json
+
+__version__ = "0.2.0"
+
+API_ENDPOINT = "https://api.warrant.dev"
+API_VERSION = "/v1"
+
+class WarrantException(Exception):
+    def __init__(self, msg, status_code=-1):
+        if status_code == -1:
+            message = 'Warrant error: ' + msg
+        else:
+            message = f"Warrant error: {status_code} " + msg
+        super().__init__(message)
+
+class User(object):
+    def __init__(self, object_type, object_id, relation):
+        self.objectType = object_type
+        self.objectId = object_id
+        self.relation = relation
+
+class Warrant(object):
+    def __init__(self, api_key):
+        self._apiKey = api_key
+
+    def _make_post_request(self, uri, json={}):
+        headers = { "Authorization": "ApiKey " + self._apiKey }
+        resp = requests.post(url = API_ENDPOINT+API_VERSION+uri, headers = headers, json = json)
+        if resp.status_code == 200:
+            return resp.json()
+        else:
+            raise WarrantException(msg=resp.text, status_code=resp.status_code)
+
+    def create_user(self, user_id=""):
+        if user_id == "":
+            payload = {}
+        else:
+            payload = { "userId": user_id }
+        json = self._make_post_request(uri="/users", json=payload)
+        return json['userId']
+
+    def create_session(self, user_id):
+        if user_id == "":
+            raise WarrantException(msg="Invalid userId provided")
+        json = self._make_post_request(uri="/users/"+user_id+"/sessions")
+        return json['token']
+
+    def create_warrant(self, object_type, object_id, relation, user):
+        if object_type == "" or object_id == "" or relation == "":
+            raise WarrantException(msg="Invalid object_type, object_id and/or relation")
+        payload = {
+            "objectType": object_type,
+            "objectId": object_id,
+            "relation": relation
+        }
+        if isinstance(user, str):
+            payload["user"] = { "userId": user }
+        elif isinstance(user, User):
+            payload["user"] = json.dumps(user.__dict__)
+        else:
+            raise WarrantException(msg="Invalid type for \'user\'. Must be of type User or str")
+        resp = self._make_post_request(uri="/warrants", json=payload)
+        return resp['id']
+
+    def is_authorized(self, object_type, object_id, relation, user_to_check):
+        if object_type == "" or object_id == "" or relation == "":
+            raise WarrantException(msg="Invalid object_type, object_id and/or relation")
+        payload = {
+            "objectType": object_type,
+            "objectId": object_id,
+            "relation": relation
+        }
+        if isinstance(user_to_check, str):
+            payload["user"] = { "userId": user_to_check }
+        elif isinstance(user_to_check, User):
+            payload["user"] = json.dumps(user_to_check.__dict__)
+        else:
+            raise WarrantException(msg="Invalid type for \'user_to_check\'. Must be of type User or str")
+        headers = { "Authorization": "ApiKey " + self._apiKey }
+        resp = requests.post(url = API_ENDPOINT+API_VERSION+"/authorize", headers = headers, json=payload)
+        if resp.status_code == 200:
+            return True
+        elif resp.status_code == 401:
+            return False
+        else:
+            raise WarrantException(msg=resp.text, status_code=resp.status_code)