# basic-rest-endpoint
For a REST API, use the BasicRestEndpoint class.
This class allows easily creating a task for an endpoint. For example, say we want to integrate with example.com's
indicator API. Here is a table of what their API looks like.
Endpoint HTTP Method URL Parameters
https://example.com/api/indicator GET indicator_id (int)
https://example.com/api/indicator POST/PATCH indicator_name, indicator_value
https://example.com/api/indicator/{id} DELETE (none)
So we have 3 endpoints, each with different HTTP Methods and parameters, but the same base URL.
We can create a superclass for this integration, let's call it ExampleIntegration
```python
from basic_rest_endpoint import BasicRestEndpoint
class ExampleIntegration(BasicRestEndpoint):
def __init__(self, context):
super(ExampleIntegration, self).__init__(context.asset["host"]
# raise_for_status=False # If we wanted to supress non-200 http codes being raised, set this to False
)
```
## Basic Request
To make a request using this library, you just use `self.request(<method>, <endpoint>, **kwargs)`
Where `<method>` is an HTTP method, `<endpoint>` is the URL relative to the host, and `**kwargs`
are optional params to pass into the requests.request(...) call
## Basic GET Example
Now we create a task, say for the GET /api/indicator endpoint
```python
from sw_example import ExampleIntegration # Import our superclass from above
class SwMain(ExampleIntegration):
endpoint = "/api/indicator"
method = "GET"
def __init__(self, context):
super(SwMain, self).__init__(context)
self.kwargs['params'] = {"indicator_id": context.inputs["indicator_id"]} # Get indicator from inputs
```
But we didn't actually make a request here! It is all handled by the BasicRestEndpoint superclass.
The params kwarg is passed into self.request which is used to create the full url with self.host that
ends up like https://example.com/api/indicator?indicator_id=<id>
## Basic POST Example
But what if the data required from the API isn't in the URL params? And what if the data returned from
wthe API isn't suitable for just returning, or needs some parsing?
Let's take a look at the second endpoint, the POST /api/indicator.
```python
from sw_example import ExampleIntegration # Import our superclass from above
class SwMain(ExampleIntegration):
endpoint = "/api/indicator"
method = "POST"
def parse_response(self, response):
data = response.json() # Basically json.loads(response.text)
return data["data"]
def __init__(self, context):
super(SwMain, self).__init__(context)
self.kwargs['json'] = {
"indicator_name": context.inputs["indicator_name"], # Get indicator from inputs
"indicator_value": context.inputs["indicator_value"]
}
}
```
This time, the data is passed in under the `json` parameter to requests which automatically formats our data for
us in the POST body. If the data were non-json, we would use `data` instead. We also parsed out the data returned,
only returning the JSON under the `data` key.
## Basic DELETE Example
Similarly to a variable request method, we can have a variable URL. This is quite trivial
```python
class SwMain(ExampleIntegration):
method = "DELETE"
def __init__(self, context):
super(SwMain, self).__init__(context)
self.endpoint = f"/api/indicator/{context.inputs['iid']}"
```
## Authentication
But what if example.com required authentication to make those calls?
There are options for these authentication methods, Basic Auth, Header Auth, Param Auth, and Custom Auth.
### Basic Auth
```python
from basic_rest_endpoint import BasicRestEndpoint
class ExampleIntegration(BasicRestEndpoint):
def __init__(self, context):
super(ExampleIntegration, self).__init__(
host=context.asset["host"],
auth=(context.asset["username"], context.asset["password"])
)
```
This auth is handled by requests directly, and automatically parses it out and inserts it into the headers for us
### Header Auth
```python
from basic_rest_endpoint import BasicRestEndpoint, HeaderAuth
class ExampleIntegration(BasicRestEndpoint):
def __init__(self, context):
super(ExampleIntegration, self).__init__(
host=context.asset["host"],
auth=HeaderAuth({"X-api-key": context.asset["api_key"]})
)
```
This auth is when an API requires a certain header to be sent in each request
### Param Auth
```python
from basic_rest_endpoint import BasicRestEndpoint, ParamAuth
class ExampleIntegration(BasicRestEndpoint):
def __init__(self, context):
super(ExampleIntegration, self).__init__(
host=context.asset["host"],
auth=ParamAuth({"username": context.asset["username"], "password" context.asset["password"]})
)
``````
This auth is used when the URL contains the authenticating information, like
https://example.com/api/indicator?indicator_id=<id>&username=<username>&password=<password>
## Polling Requests
Sometimes an API will return a status that indicates that they are still processing your request,
and you will need to send requests until the processing is complete. We can use the polling request here.
```python
# def poll_request(self, method, endpoint, step=5, timeout=60, poll_func=None, **kwargs):
# By default the polling stops if you receive a 200
# Poll /my/endpoint with default settings
self.poll_request("GET", "/my/endpoint")
# Poll /my/endpoint every 5 seconds, giving up after 20 seconds
self.poll_request("GET", "/my/endpoint", step=5, timeout=60)
# Custom polling function to check if the json returned says it's finished
def my_poll_func(poll_method, poll_endpoint, poll_kwargs):
result = self.request(poll_method, poll_endpoint, **poll_kwargs)
if r.json()["status"].lower() == "done":
return result # Return the final response
else:
return False # If what we return is falsey, then it will continue to poll
self.poll_request("GET" "/my/endpoint", poll_func=my_poll_func)
```
## Basic Pagination
An API may return a single page in a list of results of pages. To make this easy to process,
inherit from BasicPaginationEndpoint and implement the following functions
```python
from basic_rest_endpoint import BasicRestPaginationEndpoint
def MyIntegration(BasicRestPaginationEndpoint):
def __init__(self, context):
# Same init as BasicRestEndpoint, excluding in example
def get_next_page(self, response):
data = response.json()
if "next" in data:
return data["next"] # Return the URL for the next call
else:
return None # If this function returns None, all pages have been seen
def parse_response(self, response):
data = response.json()
data.pop("next", None) # Remove useless keys/clean data of each response here
return data
def combine_responses(self, results):
# Results is a list of parsed responses, from self.parse_response
all_data = []
for result in results:
all_data.extend(result) # Use .extend to take [1,2,3] + [4,5] -> [1,2,3,4,5]
return all_data
```
## Link Headers Pagination
Some (very few) APIs implement a standard called "Link Headers" which makes pagination very easy.
This implementation is completely done so all you have to do is implement combine_responses
```python
from basic_rest_endpoint import LinkHeadersPaginationEndpoint
def MyIntegration(LinkHeadersPaginationEndpoint):
def __init__(self, context):
# Same init as BasicRestEndpoint, excluding in example
def combine_responses(self, results):
# do parsing here
```
## Asset Parser
The `asset_parser` function is used to split the incoming Context object into a super() call for BasicRestEndpoint
In the following example, the Context object is parsed, and with auth set to "basic" the username and password are automatically set up for Basic HTTP auth.
```python
from basic_rest_endpoint import BasicRestEndpoint, asset_parser
class MyIntegration(BasicRestEndpoint):
def __init__(self, context):
super(MyIntegration, self).__init__(**asset_parser(context, auth="basic"))
class Context(object):
asset = {
"host": "abc.com",
"username": "bb",
"password": "cc",
"verify_ssl": False,
"http_proxy": None
}
```
Raw data
{
"_id": null,
"home_page": "https://github.com/swimlane/basic-rest-endpoint",
"name": "basic-rest-endpoint",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.6",
"maintainer_email": null,
"keywords": null,
"author": "Swimlane",
"author_email": "info@swimlane.com",
"download_url": "https://files.pythonhosted.org/packages/3c/55/d6b289547417d09b8b98e20faa79f1fb3df3b0c2bd94ca00c533fb470173/basic_rest_endpoint-1.2.0.tar.gz",
"platform": null,
"description": "# basic-rest-endpoint\n\nFor a REST API, use the BasicRestEndpoint class.\n\nThis class allows easily creating a task for an endpoint. For example, say we want to integrate with example.com's\nindicator API. Here is a table of what their API looks like.\n\nEndpoint\tHTTP Method\tURL Parameters\nhttps://example.com/api/indicator\tGET\tindicator_id (int)\nhttps://example.com/api/indicator\tPOST/PATCH\tindicator_name, indicator_value\nhttps://example.com/api/indicator/{id}\tDELETE\t(none)\nSo we have 3 endpoints, each with different HTTP Methods and parameters, but the same base URL.\nWe can create a superclass for this integration, let's call it ExampleIntegration\n\n```python\nfrom basic_rest_endpoint import BasicRestEndpoint\n\n\nclass ExampleIntegration(BasicRestEndpoint):\n def __init__(self, context):\n super(ExampleIntegration, self).__init__(context.asset[\"host\"]\n # raise_for_status=False # If we wanted to supress non-200 http codes being raised, set this to False\n )\n```\n## Basic Request\nTo make a request using this library, you just use `self.request(<method>, <endpoint>, **kwargs)`\nWhere `<method>` is an HTTP method, `<endpoint>` is the URL relative to the host, and `**kwargs`\nare optional params to pass into the requests.request(...) call\n\n## Basic GET Example\nNow we create a task, say for the GET /api/indicator endpoint\n```python\nfrom sw_example import ExampleIntegration # Import our superclass from above\n\nclass SwMain(ExampleIntegration):\n endpoint = \"/api/indicator\"\n method = \"GET\"\n\n def __init__(self, context):\n super(SwMain, self).__init__(context)\n self.kwargs['params'] = {\"indicator_id\": context.inputs[\"indicator_id\"]} # Get indicator from inputs\n```\nBut we didn't actually make a request here! It is all handled by the BasicRestEndpoint superclass.\nThe params kwarg is passed into self.request which is used to create the full url with self.host that\nends up like https://example.com/api/indicator?indicator_id=<id>\n\n## Basic POST Example\nBut what if the data required from the API isn't in the URL params? And what if the data returned from\nwthe API isn't suitable for just returning, or needs some parsing?\n\nLet's take a look at the second endpoint, the POST /api/indicator.\n```python\nfrom sw_example import ExampleIntegration # Import our superclass from above\n\n\nclass SwMain(ExampleIntegration):\n endpoint = \"/api/indicator\"\n method = \"POST\"\n\n def parse_response(self, response):\n data = response.json() # Basically json.loads(response.text)\n return data[\"data\"]\n\n def __init__(self, context):\n super(SwMain, self).__init__(context)\n self.kwargs['json'] = {\n \"indicator_name\": context.inputs[\"indicator_name\"], # Get indicator from inputs\n \"indicator_value\": context.inputs[\"indicator_value\"]\n }\n}\n```\nThis time, the data is passed in under the `json` parameter to requests which automatically formats our data for\nus in the POST body. If the data were non-json, we would use `data` instead. We also parsed out the data returned,\nonly returning the JSON under the `data` key.\n\n## Basic DELETE Example\nSimilarly to a variable request method, we can have a variable URL. This is quite trivial\n```python\nclass SwMain(ExampleIntegration):\n method = \"DELETE\"\n\n def __init__(self, context):\n super(SwMain, self).__init__(context)\n self.endpoint = f\"/api/indicator/{context.inputs['iid']}\"\n```\n\n## Authentication\nBut what if example.com required authentication to make those calls?\nThere are options for these authentication methods, Basic Auth, Header Auth, Param Auth, and Custom Auth.\n\n### Basic Auth\n```python\nfrom basic_rest_endpoint import BasicRestEndpoint\n\nclass ExampleIntegration(BasicRestEndpoint):\n def __init__(self, context):\n super(ExampleIntegration, self).__init__(\n host=context.asset[\"host\"],\n auth=(context.asset[\"username\"], context.asset[\"password\"])\n )\n```\nThis auth is handled by requests directly, and automatically parses it out and inserts it into the headers for us\n\n### Header Auth\n```python\nfrom basic_rest_endpoint import BasicRestEndpoint, HeaderAuth\n\nclass ExampleIntegration(BasicRestEndpoint):\n def __init__(self, context):\n super(ExampleIntegration, self).__init__(\n host=context.asset[\"host\"],\n auth=HeaderAuth({\"X-api-key\": context.asset[\"api_key\"]})\n )\n```\nThis auth is when an API requires a certain header to be sent in each request\n\n### Param Auth\n```python\nfrom basic_rest_endpoint import BasicRestEndpoint, ParamAuth\n\nclass ExampleIntegration(BasicRestEndpoint):\n def __init__(self, context):\n super(ExampleIntegration, self).__init__(\n host=context.asset[\"host\"],\n auth=ParamAuth({\"username\": context.asset[\"username\"], \"password\" context.asset[\"password\"]})\n )\n``````\nThis auth is used when the URL contains the authenticating information, like \nhttps://example.com/api/indicator?indicator_id=<id>&username=<username>&password=<password>\n\n## Polling Requests\nSometimes an API will return a status that indicates that they are still processing your request, \nand you will need to send requests until the processing is complete. We can use the polling request here.\n```python\n# def poll_request(self, method, endpoint, step=5, timeout=60, poll_func=None, **kwargs):\n\n# By default the polling stops if you receive a 200\n# Poll /my/endpoint with default settings\nself.poll_request(\"GET\", \"/my/endpoint\")\n\n# Poll /my/endpoint every 5 seconds, giving up after 20 seconds\nself.poll_request(\"GET\", \"/my/endpoint\", step=5, timeout=60)\n\n# Custom polling function to check if the json returned says it's finished\ndef my_poll_func(poll_method, poll_endpoint, poll_kwargs):\n result = self.request(poll_method, poll_endpoint, **poll_kwargs)\n if r.json()[\"status\"].lower() == \"done\":\n return result # Return the final response\n else:\n return False # If what we return is falsey, then it will continue to poll\n\nself.poll_request(\"GET\" \"/my/endpoint\", poll_func=my_poll_func)\n```\n\n## Basic Pagination\nAn API may return a single page in a list of results of pages. To make this easy to process,\ninherit from BasicPaginationEndpoint and implement the following functions\n```python\nfrom basic_rest_endpoint import BasicRestPaginationEndpoint\n\ndef MyIntegration(BasicRestPaginationEndpoint):\n def __init__(self, context):\n # Same init as BasicRestEndpoint, excluding in example\n\n def get_next_page(self, response):\n data = response.json()\n if \"next\" in data:\n return data[\"next\"] # Return the URL for the next call\n else:\n return None # If this function returns None, all pages have been seen\n\n def parse_response(self, response):\n data = response.json()\n data.pop(\"next\", None) # Remove useless keys/clean data of each response here\n return data\n\n def combine_responses(self, results):\n # Results is a list of parsed responses, from self.parse_response\n all_data = []\n for result in results:\n all_data.extend(result) # Use .extend to take [1,2,3] + [4,5] -> [1,2,3,4,5]\n\n return all_data\n```\n\n## Link Headers Pagination\nSome (very few) APIs implement a standard called \"Link Headers\" which makes pagination very easy.\nThis implementation is completely done so all you have to do is implement combine_responses\n```python\nfrom basic_rest_endpoint import LinkHeadersPaginationEndpoint\n\ndef MyIntegration(LinkHeadersPaginationEndpoint):\n def __init__(self, context):\n # Same init as BasicRestEndpoint, excluding in example\n\n def combine_responses(self, results):\n # do parsing here\n```\n\n## Asset Parser\nThe `asset_parser` function is used to split the incoming Context object into a super() call for BasicRestEndpoint\n\nIn the following example, the Context object is parsed, and with auth set to \"basic\" the username and password are automatically set up for Basic HTTP auth.\n```python\nfrom basic_rest_endpoint import BasicRestEndpoint, asset_parser\n\n\nclass MyIntegration(BasicRestEndpoint):\n def __init__(self, context):\n super(MyIntegration, self).__init__(**asset_parser(context, auth=\"basic\"))\n\n\nclass Context(object):\n asset = {\n \"host\": \"abc.com\",\n \"username\": \"bb\",\n \"password\": \"cc\",\n \"verify_ssl\": False,\n \"http_proxy\": None\n }\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "A package that facilitates standard rest API flows.",
"version": "1.2.0",
"project_urls": {
"Bug Tracker": "https://github.com/swimlane/basic-rest-endpoint/issues",
"Homepage": "https://github.com/swimlane/basic-rest-endpoint"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "23641e182a6c2f7b61ea9f5442f44b2d579b38904a9510aee26180ac6054dcc3",
"md5": "46f22c8462fc14abb99926ba269387be",
"sha256": "fabb83232acb2c469b6e665b0e5f999329e51c2c04c4827ec787d3ae68f22d14"
},
"downloads": -1,
"filename": "basic_rest_endpoint-1.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "46f22c8462fc14abb99926ba269387be",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.6",
"size": 10358,
"upload_time": "2024-07-03T17:26:01",
"upload_time_iso_8601": "2024-07-03T17:26:01.085746Z",
"url": "https://files.pythonhosted.org/packages/23/64/1e182a6c2f7b61ea9f5442f44b2d579b38904a9510aee26180ac6054dcc3/basic_rest_endpoint-1.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "3c55d6b289547417d09b8b98e20faa79f1fb3df3b0c2bd94ca00c533fb470173",
"md5": "42b672b34569d85b5828da223a5d4374",
"sha256": "678d117ec1e4a8274616a03928782fac614a09f08bbb3bf3fba5e9038c16977f"
},
"downloads": -1,
"filename": "basic_rest_endpoint-1.2.0.tar.gz",
"has_sig": false,
"md5_digest": "42b672b34569d85b5828da223a5d4374",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.6",
"size": 12534,
"upload_time": "2024-07-03T17:26:03",
"upload_time_iso_8601": "2024-07-03T17:26:03.137081Z",
"url": "https://files.pythonhosted.org/packages/3c/55/d6b289547417d09b8b98e20faa79f1fb3df3b0c2bd94ca00c533fb470173/basic_rest_endpoint-1.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-07-03 17:26:03",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "swimlane",
"github_project": "basic-rest-endpoint",
"github_not_found": true,
"lcname": "basic-rest-endpoint"
}
JSON
