kennethreitz/neon-api-python
1
0
Fork 0
mirror of https://github.com/kennethreitz/neon-api-python.git synced 2026-06-05 22:50:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove unnecessary code and refactor API key revocation method

Browse Source
This commit is contained in:
Kenneth Reitz
2024-01-18 13:34:02 -05:00
parent 2c3e01bf19
commit aa4d4afa84
10 changed files with 2 additions and 2324 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__first_attempt/__init__.py
-2
View File
@@ -1,2 +0,0 @@
# from .client import NeonClient
from .simple import NeonClient
__first_attempt/__version__.py
-1
View File
@@ -1 +0,0 @@
__version__ = "0.1.0"
__first_attempt/client.py
-219
View File
@@ -1,219 +0,0 @@
from .http_client import Neon_API_V2
from .resources import ResourceCollection
from . import models
class ItemView:
"""A view into a single item."""
def __init__(self, item, key_id=None):
self._item = item
self._key_id = key_id
@property
def item(self):
if self._key_id:
return getattr(self._item, self._key_id)
return self._item
def __getattr__(self, name):
return getattr(self.item, name)
def __setattr__(self, name, value):
if name == "_item":
return super().__setattr__(name, value)
return setattr(self.item, name, value)
def __str__(self):
return str(self.item)
def __repr__(self):
return repr(self.item)
def __eq__(self, other):
return self.item == other
def __ne__(self, other):
return self.item != other
class CollectionView:
"""A view into a collection of items."""
def __init__(self, collection, key_ids=None, collection_id=None):
self.pagination = None
if not key_ids:
key_ids = []
self._key_ids = key_ids
if collection_id:
self._collection = getattr(collection, collection_id)
try:
self.pagination = collection.pagination
except AttributeError:
pass
else:
self._collection = collection
def __iter__(self):
return iter(self._collection)
def __getitem__(self, key):
for k in self._key_ids:
for item in self._collection:
if str(getattr(item, k)) == str(key):
return item
return self._collection[key]
def __len__(self):
return len(self._collection)
def __repr__(self):
return repr(self._collection)
def __str__(self):
return repr(self._collection)
def __contains__(self, item):
return item in self._collection
def __eq__(self, other):
return self._collection == other
def __ne__(self, other):
return self._collection != other
class NeonClient:
def __init__(self, api_key: str, **kwargs):
self.api = Neon_API_V2(api_key, **kwargs)
self.resources = ResourceCollection(self.api)
def me(self):
return self.resources.users.get_current_user_info()
def api_keys(self):
"""Get a list of API keys."""
return CollectionView(self.resources.api_keys.get_list(), key_ids=["id"])
def projects(self, shared=False, **kwargs):
"""Get a list of projects."""
a = self.resources.projects.get_list(shared=shared, **kwargs)
print(a)
exit()
return CollectionView(
self.resources.projects.get_list(shared=shared, **kwargs),
key_ids=["id", "name"],
collection_id="projects",
)
def project(self, project_id: str, **kwargs):
"""Get a single project."""
return ItemView(
self.resources.projects.get(project_id, **kwargs), key_id="project"
)
def project_create(self, **kwargs):
return ItemView(self.resources.projects.create(**kwargs), key_id="project")
def project_delete(self, project_id: str, **kwargs):
return ItemView(
self.resources.projects.delete(project_id, **kwargs), key_id="projects"
)
def databases(self, project_id: str, branch_id: str, **kwargs):
return CollectionView(
self.resources.databases.get_list(project_id, branch_id, **kwargs),
key_ids=["id"],
collection_id="databases",
)
def database(self, project_id: str, database_id: str, **kwargs):
return ItemView(
self.resources.databases.get(project_id, database_id, **kwargs),
key_id="database",
)
def branches(self, project_id: str, **kwargs):
return CollectionView(
self.resources.branches.get_list(project_id, **kwargs),
key_ids=["id", "name"],
collection_id="branches",
)
def branch(self, project_id: str, branch_id: str):
return ItemView(
self.resources.branches.get(project_id, branch_id),
key_id="branch",
)
def branch_create(self, project_id: str, **kwargs):
return ItemView(
self.resources.branches.create(project_id, **kwargs),
key_id="branch",
)
def branch_delete(self, project_id: str, branch_id: str, **kwargs):
return ItemView(
self.resources.branches.delete(project_id, branch_id, **kwargs),
key_id="branch",
)
def branch_update(self, project_id: str, branch_id: str, **kwargs):
# TODO: untested.
return ItemView(
self.resources.branches.update(project_id, branch_id, **kwargs),
key_id="branch",
)
def branch_rename(self, project_id: str, branch_id: str, **kwargs):
# TODO: untested.
return ItemView(
self.resources.branches.rename(project_id, branch_id, **kwargs),
key_id="branch",
)
def branch_add_compute(self, project_id: str, branch_id: str, **kwargs):
# TODO: untested.
return ItemView(
self.resources.branches.add_compute(project_id, branch_id, **kwargs),
key_id="branch",
)
def branch_remove_compute(self, project_id: str, branch_id: str, **kwargs):
# TODO: untested.
return ItemView(
self.resources.branches.remove_compute(project_id, branch_id, **kwargs),
key_id="branch",
)
def branch_set_primary(self, project_id: str, branch_id: str, **kwargs):
# TODO: untested.
return ItemView(
self.resources.branches.set_primary(project_id, branch_id, **kwargs),
key_id="branch",
)
def get_connection_string(self, project_id: str, branch_id: str, database_id: str):
# TODO: implement this.
return self.resources.databases.get_connection_string(
project_id,
branch_id,
database_id,
)
# def branch_create(self, project_id: str, **kwargs):
# return ItemView(
# self.resources.branches.create(project_id, **kwargs),
# key_id="branch",
# )
__first_attempt/exceptions.py
-13
View File
@@ -1,13 +0,0 @@
from requests.exceptions import HTTPError
class NeonClientException(HTTPError):
"""Base exception class for all exceptions raised by the Neon Client."""
pass
class NeonClientAuthenticationException(NeonClientException):
"""Exception raised when authentication fails."""
pass
__first_attempt/http_client.py
-87
View File
@@ -1,87 +0,0 @@
from typing import List
import requests
from pydantic import BaseModel
from .models import *
from .exceptions import NeonClientException
from .__version__ import __version__
class Neon_API_V2:
def __init__(
self, api_key: str, *, base_url: str = "https://console.neon.tech/api/v2/"
):
self.api_key = api_key
self.base_url = base_url
self.session = requests.Session()
self.user_agent = f"neon-client/{__version__}"
def request(
self,
method: str,
path: str,
*,
response_model: BaseModel = None,
response_is_array=False,
check_status_code=True,
_debug=False,
**kwargs,
):
"""
Sends an HTTP request to the specified path using the specified method.
Args:
method (str): The HTTP method to use for the request.
path (str): The path to send the request to.
response_model (BaseModel, optional): The model to deserialize the response into. Defaults to None.
response_is_array (bool or str, optional): Indicates whether the response is a list of items. If a string is provided, it is used as the key to access the list in the response JSON. Defaults to False.
check_status_code (bool, optional): Indicates whether to check the status code of the response and raise an exception if it is not successful. Defaults to True.
**kwargs: Additional keyword arguments to be passed to the request.
Returns:
The deserialized response if a response model is provided, otherwise the raw response object.
"""
# Set HTTP headers for outgoing requests.
headers = kwargs.pop("headers", {})
headers["Authorization"] = f"Bearer {self.api_key}"
headers["Accept"] = "application/json"
headers["Content-Type"] = "application/json"
headers["User-Agent"] = self.user_agent
r = self.session.request(
method, self.base_url + path, headers=headers, **kwargs
)
if check_status_code:
# TODO: add custom exception classes here.
try:
r.raise_for_status()
except requests.exceptions.HTTPError:
raise NeonClientException(r.text)
if response_model:
if response_is_array:
# Shortcut for when the response is a list of items.
if type(response_is_array) == "str":
response_parsed = [
response_model.model_construct(**item)
for item in r.json()[response_is_array]
]
elif response_is_array == True:
response_parsed = [
response_model.model_construct(**item) for item in r.json()
]
else:
response_parsed = response_model.model_construct(**r.json())
if _debug:
print(r.json())
return response_parsed
else:
return r
def url_join(self, *args):
"""Join the specified path components into a URL."""
return "/".join(args)
__first_attempt/models.py
-1452
View File
File diff suppressed because it is too large Load Diff
__first_attempt/resources.py
-356
View File
@@ -1,356 +0,0 @@
from typing import List
from fastapi.encoders import jsonable_encoder
from .http_client import Neon_API_V2
from . import models
from .utils import validate_obj_model
class PagedOperationsResponse(models.OperationsResponse, models.PaginationResponse):
"""A response containing a list of operations and pagination information."""
pass
class PagedProjectsResponse(models.ProjectsResponse, models.PaginationResponse):
"""A response containing a list of projects and pagination information."""
pass
class Resource:
base_path = None
def __init__(self, api: Neon_API_V2):
self.api = api
class UserResource(Resource):
"""A resource for interacting with users."""
base_path = "users"
def get_current_user_info(self):
"""Get information about the user."""
return self.api.request(
method="GET",
path=self.api.url_join(self.base_path, "me"),
response_model=models.CurrentUserInfoResponse,
)
class APIKeyResource(Resource):
"""A resource for interacting with API keys."""
base_path = "api_keys"
def get_list(self):
"""Get a list of API keys."""
return self.api.request(
method="GET",
path=self.base_path,
response_model=models.ApiKeysListResponseItem,
response_is_array=True,
)
def create(self, key_name: str):
"""Create a new API key."""
return self.api.request(
method="POST",
path=self.base_path,
json=models.ApiKeyCreateRequest(key_name=key_name).model_dump(),
response_model=models.ApiKeyCreateResponse,
)
def revoke(self, key_id: str):
"""Revoke an API key."""
return self.api.request(
method="DELETE",
path=self.api.url_join(self.base_path, str(key_id)),
response_model=models.ApiKeyRevokeResponse,
)
class ProjectResource(Resource):
"""A resource for interacting with projects."""
base_path = "projects"
def get_list(
self,
*,
cursor: int | None = None,
limit: int | None = None,
shared: bool = False,
):
"""Get a list of projects."""
project_params = {}
if cursor is not None:
project_params["cursor"] = cursor
if limit is not None:
project_params["limit"] = limit
return self.api.request(
method="GET",
path=(
self.api.url_join(self.base_path, "shared")
if shared
else self.base_path
),
params=project_params,
response_model=PagedProjectsResponse,
)
def get(self, project_id: str):
"""Get a project."""
return self.api.request(
method="GET",
path=self.api.url_join(self.base_path, project_id),
response_model=models.ProjectResponse,
)
def create(self, **kwargs):
"""Create a new project."""
return self.api.request(
method="POST",
path=self.base_path,
json=kwargs,
response_model=models.ProjectResponse,
)
def update(self, project: models.Project):
"""Update a project."""
payload = models.ProjectUpdateRequest(project=project.model_dump())
return self.api.request(
method="PATCH",
path=self.api.url_join(self.base_path, project.id),
json={"project": payload.model_dump()},
response_model=models.ProjectResponse,
)
def delete(self, project_id: str):
"""Delete a project."""
return self.api.request(
method="DELETE",
path=self.api.url_join(self.base_path, project_id),
response_model=models.ProjectResponse,
)
class DatabaseResource(Resource):
base_path = "databases"
def _extract_database(self, obj):
"""Extract a database from the specified object."""
assert isinstance(
obj,
(
models.DatabaseCreateRequest,
models.Database1,
models.Database2,
models.Database,
),
)
# Object mappings.
if isinstance(obj, models.DatabaseCreateRequest):
obj = obj.database.model_dump()
if isinstance(obj, models.Database1):
obj = obj.database.model_dump()
if isinstance(obj, models.Database2):
obj = obj.database.model_dump()
if isinstance(obj, models.Database):
obj = obj.model_dump()
return obj
def get_list(self, project_id: str, branch_id: str):
"""Get a list of databases.
See also: https://api-docs.neon.tech/reference/listprojectbranchdatabases
"""
return self.api.request(
method="GET",
path=self.api.url_join(
"projects", project_id, "branches", branch_id, "databases"
),
response_model=DatabasesResponse,
)
def get(
self,
project_id: str,
branch_id: str,
database_name: str,
):
"""Get a database."""
return self.api.request(
method="GET",
path=self.api.url_join(
"projects",
project_id,
"branches",
branch_id,
"databases",
database_name,
),
response_model=DatabaseResponse,
)
def create(
self,
project_id: str,
branch_id: str,
db: models.DatabaseCreateRequest
| models.Database1
| models.Database2
| models.Database,
):
"""Create a new database."""
db = self._extract_database(db)
return self.api.request(
method="POST",
path=self.api.url_join(
"projects", project_id, "branches", branch_id, "databases"
),
json=jsonable_encoder(DatabaseCreateRequest(database=db)),
response_model=DatabaseResponse,
)
def update(
self,
project_id: str,
branch_id: str,
database_id: str,
db: models.DatabaseUpdateRequest | models.Database2,
):
"""Update a database."""
db = self._extract_database(db)
return self.api.request(
method="PATCH",
path=self.api.url_join(
"projects", project_id, "branches", branch_id, "databases", database_id
),
json=models.DatabaseUpdateRequest(database=db).model_dump(),
response_model=models.DatabaseResponse,
)
def get_connection_string(self, project_id: str, branch_id: str, database_id: str):
"""Get a database connection string."""
# TODO: this isn't correct.
return self.api.request(
method="GET",
path=self.api.url_join(
"projects",
project_id,
"branches",
branch_id,
"databases",
database_id,
"connection_string",
),
response_model=models.DatabaseConnectionStringResponse,
)
class BranchResource(Resource):
"""A resource for interacting with branches."""
path = "branches"
def get_list(self, project_id: str):
"""Get a list of branches."""
return self.api.request(
method="GET",
path=self.api.url_join("projects", project_id, "branches"),
response_model=models.BranchesResponse,
)
def get(self, project_id: str, branch_id: str):
"""Get a branch."""
return self.api.request(
method="GET",
path=self.api.url_join("projects", project_id, "branches", branch_id),
response_model=models.BranchResponse,
)
def create(self, project_id: str, request: models.BranchCreateRequest):
"""Create a new branch."""
return self.api.request(
method="POST",
path=self.api.url_join("projects", project_id, "branches"),
json=request.model_dump(),
response_model=models.BranchResponse,
)
class OperationResource(Resource):
"""A resource for interacting with operations."""
base_path = "operations"
def get_list(
self,
project_id: str,
cursor: int | None = None,
limit: int | None = None,
):
"""Get a list of operations."""
operations_params = {}
if cursor is not None:
operations_params["cursor"] = cursor
if limit is not None:
operations_params["limit"] = limit
return self.api.request(
method="GET",
path=self.api.url_join("projects", project_id, "operations"),
params=operations_params,
response_model=PagedOperationsResponse,
)
def get(self, project_id: str, operation_id: str):
"""Get an operation."""
return self.api.request(
method="GET",
path=self.api.url_join("projects", project_id, "operations", operation_id),
response_model=models.OperationResponse,
)
class ResourceCollection:
"""A collection of resources."""
def __init__(self, api: Neon_API_V2):
"""Initialize the collection."""
# Initialize resources.
self.api_keys = APIKeyResource(api)
self.users = UserResource(api)
self.projects = ProjectResource(api)
self.databases = DatabaseResource(api)
self.branches = BranchResource(api)
self.operations = OperationResource(api)
__first_attempt/simple.py
-156
View File
@@ -1,156 +0,0 @@
import os
import requests
from requests.exceptions import HTTPError
from pydantic import BaseModel
from fastapi.encoders import jsonable_encoder
from . import models
from .client import ItemView, CollectionView
from .exceptions import NeonClientException
from .__version__ import __version__
from .resources import PagedOperationsResponse, PagedProjectsResponse
from .utils import squash
NEON_ENVIRON_NAME = "NEON_API_KEY"
class NeonClient:
def __init__(
self, api_key: str, *, base_url: str = "https://console.neon.tech/api/v2/"
):
self.api_key = api_key
self.base_url = base_url
self.session = requests.Session()
self.user_agent = f"neon-client/{__version__}"
@classmethod
def from_environ(cls, environ=os.environ.copy(), **kwargs):
"""Create a new NeonClient instance from environment variables.
Args:
environ (dict, optional): The environment variables to use. Defaults to os.environ.copy().
**kwargs: Additional keyword arguments to be passed to the NeonClient constructor.
Returns:
NeonClient: The new NeonClient instance.
"""
return cls(environ[NEON_ENVIRON_NAME], **kwargs)
def url_join(self, *args):
"""Join URL path segments.
Args:
*args: The path segments to join.
Returns:
str: The joined path.
"""
return "/".join(args)
def request(
self,
method: str,
path: str,
data: dict | None = None,
*,
request_model: BaseModel | None = None,
response_model: BaseModel | None = None,
response_is_array=False,
check_status_code=True,
_debug=False,
**kwargs,
):
"""
Sends an HTTP request to the specified path using the specified method.
Args:
method (str): The HTTP method to use for the request.
path (str): The path to send the request to.
request_model (BaseModel, optional): The model to serialize the request body from. Defaults to None.
response_model (BaseModel, optional): The model to deserialize the response into. Defaults to None.
response_is_array (bool or str, optional): Indicates whether the response is a list of items. If a string is provided, it is used as the key to access the list in the response JSON. Defaults to False.
check_status_code (bool, optional): Indicates whether to check the status code of the response and raise an exception if it is not successful. Defaults to True.
**kwargs: Additional keyword arguments to be passed to the request.
Returns:
The deserialized response if a response model is provided, otherwise the raw response object.
"""
# Set HTTP headers for outgoing requests.
headers = kwargs.pop("headers", {})
headers["Authorization"] = f"Bearer {self.api_key}"
headers["Accept"] = "application/json"
headers["Content-Type"] = "application/json"
headers["User-Agent"] = self.user_agent
r = self.session.request(
method, self.base_url + path, headers=headers, **kwargs
)
if check_status_code:
# TODO: add custom exception classes here.
try:
r.raise_for_status()
except requests.exceptions.HTTPError:
raise NeonClientException(r.text)
if response_model:
if response_is_array:
# Shortcut for when the response is a list of items.
if type(response_is_array) == "str":
response_parsed = [
response_model.model_construct(**item)
for item in r.json()[response_is_array]
]
elif response_is_array is True:
response_parsed = [
response_model.model_construct(**item) for item in r.json()
]
else:
response_parsed = response_model.model_construct(**r.json())
if _debug:
print(r, r.json())
return response_parsed
else:
return r
def me(self):
return self.request(
method="GET",
path=self.url_join("users", "me"),
response_model=models.CurrentUserInfoResponse,
)
def api_keys(self):
"""Get a list of API keys."""
r = self.request(
method="GET",
path="api_keys",
response_model=models.ApiKeysListResponseItem,
response_is_array=True,
_debug=True,
)
return CollectionView(r, key_ids=["id", "name"])
def projects(
self,
*,
cursor: int | None = None,
limit: int | None = None,
shared: bool = False,
):
"""Get a list of projects."""
# Pagination support.
projects_params = squash({"cursor": cursor, "limit": limit})
r = self.request(
method="GET",
path=(self.api.url_join("projects", "shared") if shared else "projects"),
params=projects_params,
response_model=PagedProjectsResponse,
)
return CollectionView(r, key_ids=["id", "name"], collection_id="projects")
__first_attempt/utils.py
-36
View File
@@ -1,36 +0,0 @@
import inspect
from typing import List, Dict, Any, Union, Optional
from pydantic import BaseModel
def squash(obj):
new_obj = {}
for k, v in obj.items():
if isinstance(v, dict):
new_obj.update(squash(v))
else:
if v:
new_obj[k] = v
def validate_obj_model(parameter_name: str, model: BaseModel):
"""A decorator that validates the 'obj' argument against the specified model."""
def decorator(func):
def wrapper(*args, **kwargs):
# Get the 'obj' argument.
obj = kwargs.get(parameter_name)
# If the 'obj' argument is not provided, raise an exception.
if obj is None:
raise ValueError(f"Missing required argument '{parameter_name}'.")
# Validate the 'obj' argument against the specified model.
model.model_validate(obj)
# Call the wrapped function.
return func(*args, **kwargs)
return wrapper
return decorator
neon_client/client.py
+2 -2
View File
@@ -84,7 +84,7 @@ class APIKey(NeonResource):
] ]
@classmethod @classmethod
def revoke_request(cls, client, api_key): def revoke_with(cls, client, api_key):
"""Revoke an API key, via object instance.""" """Revoke an API key, via object instance."""
obj = client.request("DELETE", f"api_keys/{ api_key.obj.id }") obj = client.request("DELETE", f"api_keys/{ api_key.obj.id }")
@@ -94,7 +94,7 @@ class APIKey(NeonResource):
def revoke(self): def revoke(self):
"""Revoke the API key.""" """Revoke the API key."""
return self.revoke_request(self._client, self) return self.revoke_with(self._client, self)
class User(NeonResource): class User(NeonResource):