vultr

  9class Vultr(object):
 10    url = "https://api.vultr.com/v2"
 11
 12    def __init__(self, api_key: Optional[str] = None):
 13        """:param api_key: Vultr API Key or `VULTR_API_KEY` Environment Variable"""
 14        self.api_key = api_key or os.getenv("VULTR_API_KEY")
 15        """Provide the API key here or with the `VULTR_API_KEY` environment variable"""
 16        self._session = requests.session()
 17        if self.api_key:
 18            self._session.headers.update({"Authorization": f"Bearer {self.api_key}"})
 19
 20    def get(self, url: str, params: Optional[dict] = None) -> Any:
 21        """
 22        GET Data
 23        :param url: Request URL. Example `/instances`
 24        :param params: Query Parameters Dictionary
 25        :return: Response Data
 26        :raises: `VultrException`
 27        """
 28        return self._req("get", f"{self.url}/{url.lstrip('/')}", params=params)
 29
 30    def post(self, url: str, **kwargs) -> Any:
 31        """
 32        POST Data
 33        :param url: Request URL. Example `/instances`
 34        :param kwargs: Request Data Keyword Arguments
 35        :return: Response Data
 36        :raises: `VultrException`
 37        """
 38        return self._req("post", f"{self.url}/{url.lstrip('/')}", kwargs)
 39
 40    def patch(self, url: str, **kwargs) -> Any:
 41        """
 42        PATCH Data
 43        :param url: Request URL. Example `/instances/{instance-id}`
 44        :param kwargs: Request Data Keyword Arguments
 45        :return: Response Data
 46        :raises: `VultrException`
 47        """
 48        return self._req("patch", f"{self.url}/{url.lstrip('/')}", kwargs)
 49
 50    def put(self, url: str, **kwargs) -> Any:
 51        """
 52        PUT Data
 53        :param url: Request URL. Example `/instances/{instance-id}`
 54        :param kwargs: Request Data Keyword Arguments
 55        :return: Response Data
 56        :raises: `VultrException`
 57        """
 58        return self._req("put", f"{self.url}/{url.lstrip('/')}", kwargs)
 59
 60    def delete(self, url: str) -> None:
 61        """
 62        DELETE a Resource
 63        :param url: Request URL. Example `/instances/{instance-id}`
 64        :return: None
 65        :raises: `VultrException`
 66        """
 67        return self._req("delete", f"{self.url}/{url.lstrip('/')}")
 68
 69    def list_os(self, params: Optional[dict] = None) -> list:
 70        url = f"{self.url}/os"
 71        return self._req("get", url, params=params)["os"]
 72
 73    def list_plans(self, params: Optional[dict] = None) -> list:
 74        url = f"{self.url}/plans"
 75        return self._req("get", url, params=params)["plans"]
 76
 77    def list_regions(self, params: Optional[dict] = None) -> list:
 78        url = f"{self.url}/regions"
 79        return self._req("get", url, params=params)["regions"]
 80
 81    def list_instances(self, params: Optional[dict] = None) -> list:
 82        url = f"{self.url}/instances"
 83        return self._req("get", url, params=params)["instances"]
 84
 85    def get_instance(self, instance: Union[str, dict], params: Optional[dict] = None) -> dict:
 86        instance_id = self._get_obj_key(instance)
 87        url = f"{self.url}/instances/{instance_id}"
 88        return self._req("get", url, params=params)["instance"]
 89
 90    def create_instance(self, region: Union[str, dict], plan: Union[str, dict], **kwargs) -> dict:
 91        data = {"region": self._get_obj_key(region), "plan": self._get_obj_key(plan)}
 92        data.update(kwargs)
 93        url = f"{self.url}/instances"
 94        return self._req("post", url, data)["instance"]
 95
 96    def update_instance(self, instance: Union[str, dict], **kwargs) -> dict:
 97        instance_id = self._get_obj_key(instance)
 98        url = f"{self.url}/instances/{instance_id}"
 99        return self._req("patch", url, kwargs)["instance"]
100
101    def delete_instance(self, instance: Union[str, dict]) -> None:
102        instance_id = self._get_obj_key(instance)
103        url = f"{self.url}/instances/{instance_id}"
104        return self._req("delete", url)
105
106    def list_keys(self, params: Optional[dict] = None) -> list:
107        url = f"{self.url}/ssh-keys"
108        return self._req("get", url, params=params)["ssh_keys"]
109
110    def get_key(self, key: Union[str, dict], params: Optional[dict] = None) -> dict:
111        key_id = self._get_obj_key(key)
112        url = f"{self.url}/ssh-keys/{key_id}"
113        return self._req("get", url, params=params)["ssh_key"]
114
115    def create_key(self, name: str, key: str, **kwargs) -> dict:
116        data = {"name": name, "ssh_key": key}
117        data.update(kwargs)
118        url = f"{self.url}/ssh-keys"
119        return self._req("post", url, data)["ssh_key"]
120
121    def update_key(self, key: Union[str, dict], **kwargs) -> None:
122        key_id = self._get_obj_key(key)
123        url = f"{self.url}/ssh-keys/{key_id}"
124        return self._req("patch", url, kwargs)["ssh_key"]
125
126    def delete_key(self, key: Union[str, dict]) -> None:
127        key_id = self._get_obj_key(key)
128        url = f"{self.url}/ssh-keys/{key_id}"
129        return self._req("delete", url)
130
131    def list_scripts(self, params: Optional[dict] = None) -> list:
132        url = f"{self.url}/startup-scripts"
133        return self._req("get", url, params=params)["startup_scripts"]
134
135    def get_script(self, script: Union[str, dict], params: Optional[dict] = None) -> dict:
136        script_id = self._get_obj_key(script)
137        url = f"{self.url}/startup-scripts/{script_id}"
138        return self._req("get", url, params=params)["startup_script"]
139
140    def create_script(self, name: str, script: str, **kwargs) -> dict:
141        data = {"name": name, "script": script}
142        data.update(kwargs)
143        url = f"{self.url}/startup-scripts"
144        return self._req("post", url, data)["startup_script"]
145
146    def update_script(self, script: Union[str, dict], **kwargs) -> None:
147        script_id = self._get_obj_key(script)
148        url = f"{self.url}/startup-scripts/{script_id}"
149        return self._req("patch", url, kwargs)["startup_script"]
150
151    def delete_script(self, script: Union[str, dict]) -> None:
152        script_id = self._get_obj_key(script)
153        url = f"{self.url}/startup-scripts/{script_id}"
154        return self._req("delete", url)
155
156    def list_ipv4(self, instance: Union[str, dict], params: Optional[dict] = None) -> list:
157        instance_id = self._get_obj_key(instance)
158        url = f"{self.url}/instances/{instance_id}/ipv4"
159        return self._req("get", url, params=params)["ipv4s"]
160
161    def create_ipv4(self, instance: Union[str, dict], **kwargs) -> dict:
162        instance_id = self._get_obj_key(instance)
163        url = f"{self.url}/instances/{instance_id}/ipv4"
164        return self._req("post", url, kwargs)["ipv4"]
165
166    def delete_ipv4(self, instance: Union[str, dict]) -> None:
167        instance_id = self._get_obj_key(instance)
168        url = f"{self.url}/instances/{instance_id}/ipv4"
169        return self._req("delete", url)
170
171    @staticmethod
172    def filter_list(item_list: List[dict], value: str, key: str = "name") -> dict:
173        """
174        Helper Function to get an Item from a List of Dictionaries
175        :param item_list: List to filter
176        :param value: Value of the Key
177        :param key: Key to check for Value
178        :return: Item or {}
179        """
180        return next((d for d in item_list if str(d.get(key, "")).lower() == value.lower()), {})
181
182    @staticmethod
183    def filter_regions(regions: list, locations: list) -> list:
184        return [d for d in regions if d["id"] in locations]
185
186    @staticmethod
187    def filter_keys(keys: list, name: str) -> dict:
188        """Soft Deprecated in 0.2.0. Use `Vultr.filter_list()`"""
189        warnings.warn("Soft Deprecated in 0.2.0. Use filter_list()", PendingDeprecationWarning, stacklevel=2)
190        try:
191            return next(d for d in keys if d["name"].lower() == name.lower())
192        except StopIteration:
193            return {}
194
195    @staticmethod
196    def filter_os(os_list: list, name: str) -> dict:
197        """Soft Deprecated in 0.2.0. Use `Vultr.filter_list()`"""
198        warnings.warn("Soft Deprecated in 0.2.0. Use filter_list()", PendingDeprecationWarning, stacklevel=2)
199        try:
200            return next(d for d in os_list if d["name"].lower() == name.lower())
201        except StopIteration:
202            return {}
203
204    @staticmethod
205    def filter_scripts(scripts: list, name: str) -> dict:
206        """Soft Deprecated in 0.2.0. Use `Vultr.filter_list()`"""
207        warnings.warn("Soft Deprecated in 0.2.0. Use filter_list()", PendingDeprecationWarning, stacklevel=2)
208        try:
209            return next(d for d in scripts if d["name"].lower() == name.lower())
210        except StopIteration:
211            return {}
212
213    def _req(self, method, url, data: Any = None, params: Optional[dict] = None) -> Any:
214        r = self._session.request(method, url, params=params, json=data, timeout=10)
215        if not r.ok:
216            raise VultrException(r)
217        if r.status_code == 204:
218            return None
219        if r.headers.get("content-type") == "application/json":
220            return r.json()
221        return r.text
222
223    @staticmethod
224    def _get_obj_key(obj, key="id"):
225        if isinstance(obj, str):
226            return obj
227        elif isinstance(obj, int):
228            return str(obj)
229        elif isinstance(obj, dict):
230            if key in obj:
231                return obj[key]
232        else:
233            raise ValueError(f"Unable to parse object: {key}")

236class VultrException(Exception):
237    """Exception class for all Vultr error responses."""
238
239    def __init__(self, response: requests.Response):
240        self.status: int = response.status_code
241        """Response Status Code"""
242        try:
243            data = response.json()
244            error = data.get("error", response.text)
245        except requests.JSONDecodeError:
246            error = response.text
247        self.error: str = str(error)
248        """Error Message for 400 Codes"""
249        super().__init__(f"Error {self.status}: {self.error}")