Browse Source

Make endpoint url configurable #267 (#286)

master
Davit Tovmasyan 2 years ago
committed by Lorenzo Setale
parent
commit
1b4e087bb5
  1. 16
      README.md
  2. 2
      digitalocean/__init__.py
  3. 13
      digitalocean/baseapi.py
  4. 24
      digitalocean/tests/test_baseapi.py
  5. 3
      requirements.txt

16
README.md

@ -11,6 +11,7 @@
## Table of Contents
- [How to install](#how-to-install)
- [Configurations](#configurations)
- [Features](#features)
- [Examples](#examples)
- [Listing the droplets](#listing-the-droplets)
@ -45,6 +46,14 @@ or via sources:
**[⬆ back to top](#table-of-contents)**
## Configurations
Specify a custom provider using environment variable
export DIGITALOCEAN_END_POINT=http://example.com/
**[⬆ back to top](#table-of-contents)**
## Features
python-digitalocean support all the features provided via digitalocean.com APIs, such as:
@ -75,6 +84,13 @@ my_droplets = manager.get_all_droplets()
print(my_droplets)
```
This example shows how to specify custom provider's end point URL:
```python
import digitalocean
manager = digitalocean.Manager(token="secretspecialuniquesnowflake", end_point="http://example.com/")
```
**[⬆ back to top](#table-of-contents)**
### Listing the droplets by tags

2
digitalocean/__init__.py

@ -20,7 +20,7 @@ from .SSHKey import SSHKey
from .Kernel import Kernel
from .FloatingIP import FloatingIP
from .Volume import Volume
from .baseapi import Error, TokenError, DataReadError
from .baseapi import Error, EndPointError, TokenError, DataReadError
from .Tag import Tag
from .LoadBalancer import LoadBalancer
from .LoadBalancer import StickySesions, ForwardingRule, HealthCheck

13
digitalocean/baseapi.py

@ -38,6 +38,10 @@ class NotFoundError(Error):
pass
class EndPointError(Error):
pass
class BaseAPI(object):
"""
Basic api class for
@ -47,13 +51,20 @@ class BaseAPI(object):
def __init__(self, *args, **kwargs):
self.token = os.getenv("DIGITALOCEAN_ACCESS_TOKEN", "")
self.end_point = "https://api.digitalocean.com/v2/"
self.end_point = os.getenv("DIGITALOCEAN_END_POINT", "https://api.digitalocean.com/v2/")
self._log = logging.getLogger(__name__)
self._session = requests.Session()
for attr in kwargs.keys():
setattr(self, attr, kwargs[attr])
parsed_url = urlparse.urlparse(self.end_point)
if not parsed_url.scheme or not parsed_url.netloc:
raise EndPointError("Provided end point is not a valid URL. Please use a valid URL")
if not parsed_url.path:
self.end_point += '/'
def __getstate__(self):
state = self.__dict__.copy()

24
digitalocean/tests/test_baseapi.py

@ -1,3 +1,9 @@
import os
try:
import mock
except ImportError:
from unittest import mock
import responses
import requests
import digitalocean
@ -41,3 +47,21 @@ class TestBaseAPI(BaseTest):
self.manager._session.proxies['https'] = 'https://127.0.0.1:3128'
self.manager.get_account()
def test_custom_endpoint(self):
custom_endpoint = 'http://example.com/'
with mock.patch.dict(os.environ,
{'DIGITALOCEAN_END_POINT': custom_endpoint},
clear=True):
base_api = digitalocean.baseapi.BaseAPI()
self.assertEqual(base_api.end_point, custom_endpoint)
def test_invalid_custom_endpoint(self):
custom_endpoint = 'not a valid endpoint'
with mock.patch.dict(os.environ,
{'DIGITALOCEAN_END_POINT': custom_endpoint},
clear=True):
self.assertRaises(digitalocean.EndPointError, digitalocean.baseapi.BaseAPI)

3
requirements.txt

@ -3,4 +3,5 @@ jsonpickle
# Testing requirements
pytest
responses
responses
mock; python_version < '3.3'
Loading…
Cancel
Save