-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
4 changed files
with
142 additions
and
162 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,2 +1,74 @@ | ||
| # pyservicebinding | ||
| Kubernetes Service Binding Library for Python Applications | ||
|  | ||
|  | ||
|  | ||
| [](https://github.com/baijum/pyservicebinding/actions/workflows/python-package.yml) | ||
| > Kubernetes Service Binding Library for Python Applications | ||
| This is a Python module to retrieve bindings from a file-system created through | ||
| an implementation of [Service Binding Specification for | ||
| Kubernetes](https://github.com/k8s-service-bindings/spec). | ||
|
|
||
| You can install this package using pip: | ||
|
|
||
| ```bash | ||
| pip install pyservicebinding | ||
| ``` | ||
|
|
||
|
|
||
| The `ServiceBinding` object can be instantiated like this: | ||
| ```python | ||
| from pyservicebinding import binding | ||
| try: | ||
| sb = binding.ServiceBinding() | ||
| except binding.ServiceBindingRootMissingError as msg: | ||
| # log the error message and retry/exit | ||
| print("SERVICE_BINDING_ROOT env var not set") | ||
| ``` | ||
|
|
||
| To get bindings for a specific `type`, say `postgres`: | ||
|
|
||
| ```python | ||
| bindings_list = sb.bindings("postgres") | ||
| ``` | ||
|
|
||
| To get bindings for a specific `type`, say `mysql`, and `provider`, say `mariadb`: | ||
|
|
||
| ```python | ||
| bindings_list = sb.bindings("mysql", "mariadb") | ||
| ``` | ||
|
|
||
| To get all bindings irrespective of the `type` and `provider`: | ||
|
|
||
| ```python | ||
| bindings_list = sb.all_bindings("mariadb") | ||
| ``` | ||
|
|
||
| This is the complete API of the module: | ||
| ```python | ||
|
|
||
| class ServiceBindingRootMissingError(Exception): | ||
| pass | ||
|
|
||
|
|
||
| class ServiceBinding: | ||
|
|
||
| def __init__(self): | ||
| """ | ||
| - raise ServiceBindingRootMissingError if SERVICE_BINDING_ROOT env var not set | ||
| """ | ||
|
|
||
| def all_bindings(self) -> list[dict[str, str]]: | ||
| """Get all bindings as a list of dictionaries | ||
| - return empty list if no bindings found | ||
| """ | ||
|
|
||
| def bindings(self, _type: str, provider: typing.Optional[str] = None) -> list[dict[str, str]]: | ||
| """Get filtered bindings as a list of dictionaries | ||
| - return empty dictionary if no binding found | ||
| - filter the result with the given _type input | ||
| - if provider input is given, filter bindings using the given type and provider | ||
| """ | ||
| ``` |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,78 +1,73 @@ | ||
| import os | ||
| import typing | ||
|
|
||
| class ServiceBindingRootMissingError(KeyError): | ||
| class ServiceBindingRootMissingError(Exception): | ||
| pass | ||
|
|
||
| class DuplicateEntryError(KeyError): | ||
| pass | ||
|
|
||
| def all_bindings() -> list[dict[str, str]]: | ||
| """Get all bindings as a list of dictionaries | ||
|
|
||
| - return empty list if no bindings found | ||
| - raise ServiceBindingRootMissingError if SERVICE_BINDING_ROOT env var not set | ||
| """ | ||
| class ServiceBinding: | ||
|
|
||
| try: | ||
| root = os.environ["SERVICE_BINDING_ROOT"] | ||
| except KeyError as msg: | ||
| raise ServiceBindingRootMissingError(msg) | ||
| def __init__(self): | ||
| try: | ||
| self.root = os.environ["SERVICE_BINDING_ROOT"] | ||
| except KeyError as msg: | ||
| raise ServiceBindingRootMissingError(msg) | ||
|
|
||
| l = [] | ||
| for dirname in os.listdir(root): | ||
| b = {} | ||
| for filename in os.listdir(os.path.join(root, dirname)): | ||
| b[filename] = open(os.path.join(root, dirname, filename)).read().strip() | ||
|
|
||
| l.append(b) | ||
| def all_bindings(self) -> list[dict[str, str]]: | ||
| """Get all bindings as a list of dictionaries | ||
| return l | ||
| - return empty list if no bindings found | ||
| """ | ||
| root = self.root | ||
| l = [] | ||
| for dirname in os.listdir(root): | ||
| b = {} | ||
| for filename in os.listdir(os.path.join(root, dirname)): | ||
| b[filename] = open(os.path.join(root, dirname, filename)).read().strip() | ||
|
|
||
| def get_binding(_type: str, provider: typing.Optional[str] = None) -> dict[str, str]: | ||
| """Get binding as a dictionary for a given type and optional provider | ||
| l.append(b) | ||
|
|
||
| - return empty dictionary if no binding found | ||
| - raise DuplicateEntryError if duplicate entry found | ||
| - raise ServiceBindingRootMissingError if SERVICE_BINDING_ROOT env var not set | ||
| """ | ||
| return l | ||
|
|
||
| try: | ||
| root = os.environ["SERVICE_BINDING_ROOT"] | ||
| except KeyError as msg: | ||
| raise ServiceBindingRootMissingError(msg) | ||
| def bindings(self, _type: str, provider: typing.Optional[str] = None) -> list[dict[str, str]]: | ||
| """Get filtered bindings as a list of dictionaries | ||
| b = {} | ||
| dupcheck = [] | ||
| if provider: | ||
| for dirname in os.listdir(root): | ||
| typepath = os.path.join(root, dirname, "type") | ||
| providerpath = os.path.join(root, dirname, "provider") | ||
| if os.path.exists(typepath): | ||
| typevalue = open(typepath).read().strip() | ||
| if typevalue != _type: | ||
| continue | ||
| if os.path.exists(providerpath): | ||
| providervalue = open(providerpath).read().strip() | ||
| if providervalue != provider: | ||
| - return empty dictionary if no binding found | ||
| - filter the result with the given _type input | ||
| - if provider input is given, filter bindings using the given type and provider | ||
| """ | ||
| root = self.root | ||
| l = [] | ||
| b = {} | ||
| if provider: | ||
| for dirname in os.listdir(root): | ||
| typepath = os.path.join(root, dirname, "type") | ||
| providerpath = os.path.join(root, dirname, "provider") | ||
| if os.path.exists(typepath): | ||
| typevalue = open(typepath).read().strip() | ||
| if typevalue != _type: | ||
| continue | ||
| if os.path.exists(providerpath): | ||
| providervalue = open(providerpath).read().strip() | ||
| if providervalue != provider: | ||
| continue | ||
|
|
||
| for filename in os.listdir(os.path.join(root, dirname)): | ||
| b[filename] = open(os.path.join(root, dirname, filename)).read().strip() | ||
|
|
||
| l.append(b) | ||
| else: | ||
| for dirname in os.listdir(root): | ||
| typepath = os.path.join(root, dirname, "type") | ||
| if os.path.exists(typepath): | ||
| typevalue = open(typepath).read().strip() | ||
| if typevalue != _type: | ||
| continue | ||
| dupcheck.append(typevalue + ":" + providervalue) | ||
|
|
||
| for filename in os.listdir(os.path.join(root, dirname)): | ||
| b[filename] = open(os.path.join(root, dirname, filename)).read().strip() | ||
| else: | ||
| for dirname in os.listdir(root): | ||
| typepath = os.path.join(root, dirname, "type") | ||
| if os.path.exists(typepath): | ||
| typevalue = open(typepath).read().strip() | ||
| if typevalue != _type: | ||
| continue | ||
| dupcheck.append(typevalue) | ||
|
|
||
| for filename in os.listdir(os.path.join(root, dirname)): | ||
| b[filename] = open(os.path.join(root, dirname, filename)).read().strip() | ||
|
|
||
| if len(dupcheck) > 1 and all(x==dupcheck[0] for x in dupcheck): | ||
| raise DuplicateEntryError(dupcheck) | ||
| l.append(b) | ||
|
|
||
| return b | ||
| return l |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters