From db43b9650ea9b33dbc8c14e9c635d67ab7263cdb Mon Sep 17 00:00:00 2001 From: Palp1tate <1939311091@qq.com> Date: Wed, 6 Sep 2023 18:41:02 +0800 Subject: [PATCH] fix: refactor file structures (#66) * feat: add all missing APIs * feat: add all missing APIs * feat: add all missing APIs * fix: refactor file structures --- src/casdoor/adapter.py | 66 ++ src/casdoor/application.py | 66 ++ src/casdoor/cert.py | 67 ++ src/casdoor/enforcer.py | 66 ++ src/casdoor/group.py | 66 ++ src/casdoor/main.py | 1271 ----------------------------------- src/casdoor/model.py | 66 ++ src/casdoor/organization.py | 66 ++ src/casdoor/payment.py | 66 ++ src/casdoor/permisssion.py | 66 ++ src/casdoor/plan.py | 66 ++ src/casdoor/pricing.py | 66 ++ src/casdoor/product.py | 66 ++ src/casdoor/provider.py | 66 ++ src/casdoor/resource.py | 71 ++ src/casdoor/role.py | 66 ++ src/casdoor/session.py | 66 ++ src/casdoor/subscription.py | 67 +- src/casdoor/syncer.py | 66 ++ src/casdoor/token.py | 68 ++ src/casdoor/user.py | 90 +++ src/casdoor/webhook.py | 66 ++ 22 files changed, 1418 insertions(+), 1272 deletions(-) diff --git a/src/casdoor/adapter.py b/src/casdoor/adapter.py index 80801e9..d204a6e 100644 --- a/src/casdoor/adapter.py +++ b/src/casdoor/adapter.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Adapter: @@ -34,3 +40,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class AdapterSDK(CasdoorSDK): + def get_adapters(self) -> List[Dict]: + """ + Get the adapters from Casdoor. + + :return: a list of dicts containing adapter info + """ + url = self.endpoint + "/api/get-adapters" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + adapters = r.json() + return adapters + + def get_adapter(self, adapter_id: str) -> Dict: + """ + Get the adapter from Casdoor providing the adapter_id. + + :param adapter_id: the id of the adapter + :return: a dict that contains adapter's info + """ + url = self.endpoint + "/api/get-adapter" + params = { + "id": f"{self.org_name}/{adapter_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + adapter = r.json() + return adapter + + def modify_adapter(self, method: str, adapter: Adapter) -> Dict: + url = self.endpoint + f"/api/{method}" + adapter.owner = self.org_name + params = { + "id": f"{adapter.owner}/{adapter.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + adapter_info = json.dumps(adapter.to_dict()) + r = requests.post(url, params=params, data=adapter_info) + response = r.json() + return response + + def add_adapter(self, adapter: Adapter) -> Dict: + response = self.modify_adapter("add-adapter", adapter) + return response + + def update_adapter(self, adapter: Adapter) -> Dict: + response = self.modify_adapter("update-adapter", adapter) + return response + + def delete_adapter(self, adapter: Adapter) -> Dict: + response = self.modify_adapter("delete-adapter", adapter) + return response diff --git a/src/casdoor/application.py b/src/casdoor/application.py index 95ea077..9ac9e74 100644 --- a/src/casdoor/application.py +++ b/src/casdoor/application.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK from .organization import Organization, ThemeData from .provider import Provider @@ -96,3 +102,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class ApplicationSDK(CasdoorSDK): + def get_applications(self) -> List[Dict]: + """ + Get the applications from Casdoor. + + :return: a list of dicts containing application info + """ + url = self.endpoint + "/api/get-applications" + params = { + "owner": "admin", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + applications = r.json() + return applications + + def get_application(self, application_id: str) -> Dict: + """ + Get the application from Casdoor providing the application_id. + + :param application_id: the id of the application + :return: a dict that contains application's info + """ + url = self.endpoint + "/api/get-application" + params = { + "id": f"{self.org_name}/{application_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + application = r.json() + return application + + def modify_application(self, method: str, application: Application) -> Dict: + url = self.endpoint + f"/api/{method}" + application.owner = self.org_name + params = { + "id": f"{application.owner}/{application.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + application_info = json.dumps(application.to_dict()) + r = requests.post(url, params=params, data=application_info) + response = r.json() + return response + + def add_application(self, application: Application) -> Dict: + response = self.modify_application("add-application", application) + return response + + def update_application(self, application: Application) -> Dict: + response = self.modify_application("update-application", application) + return response + + def delete_application(self, application: Application) -> Dict: + response = self.modify_application("delete-application", application) + return response diff --git a/src/casdoor/cert.py b/src/casdoor/cert.py index ce99ba4..a4e6b66 100644 --- a/src/casdoor/cert.py +++ b/src/casdoor/cert.py @@ -12,6 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK + class Cert: def __init__(self): @@ -34,3 +41,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class CertSDK(CasdoorSDK): + def get_certs(self) -> List[Dict]: + """ + Get the certs from Casdoor. + + :return: a list of dicts containing cert info + """ + url = self.endpoint + "/api/get-certs" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + certs = r.json() + return certs + + def get_cert(self, cert_id: str) -> Dict: + """ + Get the cert from Casdoor providing the cert_id. + + :param cert_id: the id of the cert + :return: a dict that contains cert's info + """ + url = self.endpoint + "/api/get-cert" + params = { + "id": f"{self.org_name}/{cert_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + cert = r.json() + return cert + + def modify_cert(self, method: str, cert: Cert) -> Dict: + url = self.endpoint + f"/api/{method}" + cert.owner = self.org_name + params = { + "id": f"{cert.owner}/{cert.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + cert_info = json.dumps(cert.to_dict()) + r = requests.post(url, params=params, data=cert_info) + response = r.json() + return response + + def add_cert(self, cert: Cert) -> Dict: + response = self.modify_cert("add-cert", cert) + return response + + def update_cert(self, cert: Cert) -> Dict: + response = self.modify_cert("update-cert", cert) + return response + + def delete_cert(self, cert: Cert) -> Dict: + response = self.modify_cert("delete-cert", cert) + return response diff --git a/src/casdoor/enforcer.py b/src/casdoor/enforcer.py index 07b95f1..081c8e0 100644 --- a/src/casdoor/enforcer.py +++ b/src/casdoor/enforcer.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Enforcer: @@ -30,3 +36,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class EnforcerSDK(CasdoorSDK): + def get_enforcers(self) -> List[Dict]: + """ + Get the enforcers from Casdoor. + + :return: a list of dicts containing enforcer info + """ + url = self.endpoint + "/api/get-enforcers" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + enforcers = r.json() + return enforcers + + def get_enforcer(self, enforcer_id: str) -> Dict: + """ + Get the enforcer from Casdoor providing the enforcer_id. + + :param enforcer_id: the id of the enforcer + :return: a dict that contains enforcer's info + """ + url = self.endpoint + "/api/get-enforcer" + params = { + "id": f"{self.org_name}/{enforcer_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + enforcer = r.json() + return enforcer + + def modify_enforcer(self, method: str, enforcer: Enforcer) -> Dict: + url = self.endpoint + f"/api/{method}" + enforcer.owner = self.org_name + params = { + "id": f"{enforcer.owner}/{enforcer.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + enforcer_info = json.dumps(enforcer.to_dict()) + r = requests.post(url, params=params, data=enforcer_info) + response = r.json() + return response + + def add_enforcer(self, enforcer: Enforcer) -> Dict: + response = self.modify_enforcer("add-enforcer", enforcer) + return response + + def update_enforcer(self, enforcer: Enforcer) -> Dict: + response = self.modify_enforcer("update-enforcer", enforcer) + return response + + def delete_enforcer(self, enforcer: Enforcer) -> Dict: + response = self.modify_enforcer("delete-enforcer", enforcer) + return response diff --git a/src/casdoor/group.py b/src/casdoor/group.py index c0519a2..3235a51 100644 --- a/src/casdoor/group.py +++ b/src/casdoor/group.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK from .user import User @@ -38,3 +44,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class GroupSDK(CasdoorSDK): + def get_groups(self) -> List[Dict]: + """ + Get the groups from Casdoor. + + :return: a list of dicts containing group info + """ + url = self.endpoint + "/api/get-groups" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + groups = r.json() + return groups + + def get_group(self, group_id: str) -> Dict: + """ + Get the group from Casdoor providing the group_id. + + :param group_id: the id of the group + :return: a dict that contains group's info + """ + url = self.endpoint + "/api/get-group" + params = { + "id": f"{self.org_name}/{group_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + group = r.json() + return group + + def modify_group(self, method: str, group: Group) -> Dict: + url = self.endpoint + f"/api/{method}" + group.owner = self.org_name + params = { + "id": f"{group.owner}/{group.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + group_info = json.dumps(group.to_dict()) + r = requests.post(url, params=params, data=group_info) + response = r.json() + return response + + def add_group(self, group: Group) -> Dict: + response = self.modify_group("add-group", group) + return response + + def update_group(self, group: Group) -> Dict: + response = self.modify_group("update-group", group) + return response + + def delete_group(self, group: Group) -> Dict: + response = self.modify_group("delete-group", group) + return response diff --git a/src/casdoor/main.py b/src/casdoor/main.py index cf7e6c1..194b897 100644 --- a/src/casdoor/main.py +++ b/src/casdoor/main.py @@ -12,7 +12,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import json from typing import Dict, List, Optional import jwt @@ -20,28 +19,6 @@ from cryptography import x509 from cryptography.hazmat.backends import default_backend -from .adapter import Adapter -from .application import Application -from .cert import Cert -from .enforcer import Enforcer -from .group import Group -from .model import Model -from .organization import Organization -from .payment import Payment -from .permisssion import Permission -from .plan import Plan -from .pricing import Pricing -from .product import Product -from .provider import Provider -from .resource import Resource -from .role import Role -from .session import Session -from .subscription import Subscription -from .syncer import Syncer -from .token import Token -from .user import User -from .webhook import Webhook - class CasdoorSDK: def __init__( @@ -313,1251 +290,3 @@ def map_rule(rule: List[str], idx) -> Dict: raise ValueError(error_str) return enforce_results - - def get_users(self) -> List[Dict]: - """ - Get the users from Casdoor. - - :return: a list of dicts containing user info - """ - url = self.endpoint + "/api/get-users" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - users = r.json() - return users - - def get_user(self, user_id: str) -> Dict: - """ - Get the user from Casdoor providing the user_id. - - :param user_id: the id of the user - :return: a dict that contains user's info - """ - url = self.endpoint + "/api/get-user" - params = { - "id": f"{self.org_name}/{user_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - user = r.json() - return user - - def get_user_count(self, is_online: bool = None) -> int: - """ - Get the count of filtered users for an organization - :param is_online: True for online users, False for offline users, - None for all users - :return: the count of filtered users for an organization - """ - url = self.endpoint + "/api/get-user-count" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - - if is_online is None: - params["isOnline"] = "" - else: - params["isOnline"] = "1" if is_online else "0" - - r = requests.get(url, params) - count = r.json() - return count - - def modify_user(self, method: str, user: User) -> Dict: - url = self.endpoint + f"/api/{method}" - user.owner = self.org_name - params = { - "id": f"{user.owner}/{user.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - user_info = json.dumps(user.to_dict()) - r = requests.post(url, params=params, data=user_info) - response = r.json() - return response - - def add_user(self, user: User) -> Dict: - response = self.modify_user("add-user", user) - return response - - def update_user(self, user: User) -> Dict: - response = self.modify_user("update-user", user) - return response - - def delete_user(self, user: User) -> Dict: - response = self.modify_user("delete-user", user) - return response - - def get_certs(self) -> List[Dict]: - """ - Get the certs from Casdoor. - - :return: a list of dicts containing cert info - """ - url = self.endpoint + "/api/get-certs" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - certs = r.json() - return certs - - def get_cert(self, cert_id: str) -> Dict: - """ - Get the cert from Casdoor providing the cert_id. - - :param cert_id: the id of the cert - :return: a dict that contains cert's info - """ - url = self.endpoint + "/api/get-cert" - params = { - "id": f"{self.org_name}/{cert_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - cert = r.json() - return cert - - def modify_cert(self, method: str, cert: Cert) -> Dict: - url = self.endpoint + f"/api/{method}" - cert.owner = self.org_name - params = { - "id": f"{cert.owner}/{cert.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - cert_info = json.dumps(cert.to_dict()) - r = requests.post(url, params=params, data=cert_info) - response = r.json() - return response - - def add_cert(self, cert: Cert) -> Dict: - response = self.modify_cert("add-cert", cert) - return response - - def update_cert(self, cert: Cert) -> Dict: - response = self.modify_cert("update-cert", cert) - return response - - def delete_cert(self, cert: Cert) -> Dict: - response = self.modify_cert("delete-cert", cert) - return response - - def get_adapters(self) -> List[Dict]: - """ - Get the adapters from Casdoor. - - :return: a list of dicts containing adapter info - """ - url = self.endpoint + "/api/get-adapters" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - adapters = r.json() - return adapters - - def get_adapter(self, adapter_id: str) -> Dict: - """ - Get the adapter from Casdoor providing the adapter_id. - - :param adapter_id: the id of the adapter - :return: a dict that contains adapter's info - """ - url = self.endpoint + "/api/get-adapter" - params = { - "id": f"{self.org_name}/{adapter_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - adapter = r.json() - return adapter - - def modify_adapter(self, method: str, adapter: Adapter) -> Dict: - url = self.endpoint + f"/api/{method}" - adapter.owner = self.org_name - params = { - "id": f"{adapter.owner}/{adapter.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - adapter_info = json.dumps(adapter.to_dict()) - r = requests.post(url, params=params, data=adapter_info) - response = r.json() - return response - - def add_adapter(self, adapter: Adapter) -> Dict: - response = self.modify_adapter("add-adapter", adapter) - return response - - def update_adapter(self, adapter: Adapter) -> Dict: - response = self.modify_adapter("update-adapter", adapter) - return response - - def delete_adapter(self, adapter: Adapter) -> Dict: - response = self.modify_adapter("delete-adapter", adapter) - return response - - def get_groups(self) -> List[Dict]: - """ - Get the groups from Casdoor. - - :return: a list of dicts containing group info - """ - url = self.endpoint + "/api/get-groups" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - groups = r.json() - return groups - - def get_group(self, group_id: str) -> Dict: - """ - Get the group from Casdoor providing the group_id. - - :param group_id: the id of the group - :return: a dict that contains group's info - """ - url = self.endpoint + "/api/get-group" - params = { - "id": f"{self.org_name}/{group_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - group = r.json() - return group - - def modify_group(self, method: str, group: Group) -> Dict: - url = self.endpoint + f"/api/{method}" - group.owner = self.org_name - params = { - "id": f"{group.owner}/{group.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - group_info = json.dumps(group.to_dict()) - r = requests.post(url, params=params, data=group_info) - response = r.json() - return response - - def add_group(self, group: Group) -> Dict: - response = self.modify_group("add-group", group) - return response - - def update_group(self, group: Group) -> Dict: - response = self.modify_group("update-group", group) - return response - - def delete_group(self, group: Group) -> Dict: - response = self.modify_group("delete-group", group) - return response - - def get_roles(self) -> List[Dict]: - """ - Get the roles from Casdoor. - - :return: a list of dicts containing role info - """ - url = self.endpoint + "/api/get-roles" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - roles = r.json() - return roles - - def get_role(self, role_id: str) -> Dict: - """ - Get the role from Casdoor providing the role_id. - - :param role_id: the id of the role - :return: a dict that contains role's info - """ - url = self.endpoint + "/api/get-role" - params = { - "id": f"{self.org_name}/{role_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - role = r.json() - return role - - def modify_role(self, method: str, role: Role) -> Dict: - url = self.endpoint + f"/api/{method}" - role.owner = self.org_name - params = { - "id": f"{role.owner}/{role.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - role_info = json.dumps(role.to_dict()) - r = requests.post(url, params=params, data=role_info) - response = r.json() - return response - - def add_role(self, role: Role) -> Dict: - response = self.modify_role("add-role", role) - return response - - def update_role(self, role: Role) -> Dict: - response = self.modify_role("update-role", role) - return response - - def delete_role(self, role: Role) -> Dict: - response = self.modify_role("delete-role", role) - return response - - def get_organizations(self) -> List[Dict]: - """ - Get the organizations from Casdoor. - - :return: a list of dicts containing organization info - """ - url = self.endpoint + "/api/get-organizations" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - organizations = r.json() - return organizations - - def get_organization(self, organization_id: str) -> Dict: - """ - Get the organization from Casdoor providing the organization_id. - - :param organization_id: the id of the organization - :return: a dict that contains organization's info - """ - url = self.endpoint + "/api/get-organization" - params = { - "id": f"{self.org_name}/{organization_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - organization = r.json() - return organization - - def modify_organization(self, method: str, organization: Organization) -> Dict: - url = self.endpoint + f"/api/{method}" - organization.owner = self.org_name - params = { - "id": f"{organization.owner}/{organization.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - organization_info = json.dumps(organization.to_dict()) - r = requests.post(url, params=params, data=organization_info) - response = r.json() - return response - - def add_organization(self, organization: Organization) -> Dict: - response = self.modify_organization("add-organization", organization) - return response - - def update_organization(self, organization: Organization) -> Dict: - response = self.modify_organization("update-organization", organization) - return response - - def delete_organization(self, organization: Organization) -> Dict: - response = self.modify_organization("delete-organization", organization) - return response - - def get_payments(self) -> List[Dict]: - """ - Get the payments from Casdoor. - - :return: a list of dicts containing payment info - """ - url = self.endpoint + "/api/get-payments" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - payments = r.json() - return payments - - def get_payment(self, payment_id: str) -> Dict: - """ - Get the payment from Casdoor providing the payment_id. - - :param payment_id: the id of the payment - :return: a dict that contains payment's info - """ - url = self.endpoint + "/api/get-payment" - params = { - "id": f"{self.org_name}/{payment_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - payment = r.json() - return payment - - def modify_payment(self, method: str, payment: Payment) -> Dict: - url = self.endpoint + f"/api/{method}" - payment.owner = self.org_name - params = { - "id": f"{payment.owner}/{payment.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - payment_info = json.dumps(payment.to_dict()) - r = requests.post(url, params=params, data=payment_info) - response = r.json() - return response - - def add_payment(self, payment: Payment) -> Dict: - response = self.modify_payment("add-payment", payment) - return response - - def update_payment(self, payment: Payment) -> Dict: - response = self.modify_payment("update-payment", payment) - return response - - def delete_payment(self, payment: Payment) -> Dict: - response = self.modify_payment("delete-payment", payment) - return response - - def get_providers(self) -> List[Dict]: - """ - Get the providers from Casdoor. - - :return: a list of dicts containing provider info - """ - url = self.endpoint + "/api/get-providers" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - providers = r.json() - return providers - - def get_provider(self, provider_id: str) -> Dict: - """ - Get the provider from Casdoor providing the provider_id. - - :param provider_id: the id of the provider - :return: a dict that contains provider's info - """ - url = self.endpoint + "/api/get-provider" - params = { - "id": f"{self.org_name}/{provider_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - provider = r.json() - return provider - - def modify_provider(self, method: str, provider: Provider) -> Dict: - url = self.endpoint + f"/api/{method}" - provider.owner = self.org_name - params = { - "id": f"{provider.owner}/{provider.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - provider_info = json.dumps(provider.to_dict()) - r = requests.post(url, params=params, data=provider_info) - response = r.json() - return response - - def add_provider(self, provider: Provider) -> Dict: - response = self.modify_provider("add-provider", provider) - return response - - def update_provider(self, provider: Provider) -> Dict: - response = self.modify_provider("update-provider", provider) - return response - - def delete_provider(self, provider: Provider) -> Dict: - response = self.modify_provider("delete-provider", provider) - return response - - def get_applications(self) -> List[Dict]: - """ - Get the applications from Casdoor. - - :return: a list of dicts containing application info - """ - url = self.endpoint + "/api/get-applications" - params = { - "owner": "admin", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - applications = r.json() - return applications - - def get_application(self, application_id: str) -> Dict: - """ - Get the application from Casdoor providing the application_id. - - :param application_id: the id of the application - :return: a dict that contains application's info - """ - url = self.endpoint + "/api/get-application" - params = { - "id": f"{self.org_name}/{application_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - application = r.json() - return application - - def modify_application(self, method: str, application: Application) -> Dict: - url = self.endpoint + f"/api/{method}" - application.owner = self.org_name - params = { - "id": f"{application.owner}/{application.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - application_info = json.dumps(application.to_dict()) - r = requests.post(url, params=params, data=application_info) - response = r.json() - return response - - def add_application(self, application: Application) -> Dict: - response = self.modify_application("add-application", application) - return response - - def update_application(self, application: Application) -> Dict: - response = self.modify_application("update-application", application) - return response - - def delete_application(self, application: Application) -> Dict: - response = self.modify_application("delete-application", application) - return response - - def get_models(self) -> List[Dict]: - """ - Get the models from Casdoor. - - :return: a list of dicts containing model info - """ - url = self.endpoint + "/api/get-models" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - models = r.json() - return models - - def get_model(self, model_id: str) -> Dict: - """ - Get the model from Casdoor providing the model_id. - - :param model_id: the id of the model - :return: a dict that contains model's info - """ - url = self.endpoint + "/api/get-model" - params = { - "id": f"{self.org_name}/{model_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - model = r.json() - return model - - def modify_model(self, method: str, model: Model) -> Dict: - url = self.endpoint + f"/api/{method}" - model.owner = self.org_name - params = { - "id": f"{model.owner}/{model.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - model_info = json.dumps(model.to_dict()) - r = requests.post(url, params=params, data=model_info) - response = r.json() - return response - - def add_model(self, model: Model) -> Dict: - response = self.modify_model("add-model", model) - return response - - def update_model(self, model: Model) -> Dict: - response = self.modify_model("update-model", model) - return response - - def delete_model(self, model: Model) -> Dict: - response = self.modify_model("delete-model", model) - return response - - def get_plans(self) -> List[Dict]: - """ - Get the plans from Casdoor. - - :return: a list of dicts containing plan info - """ - url = self.endpoint + "/api/get-plans" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - plans = r.json() - return plans - - def get_plan(self, plan_id: str) -> Dict: - """ - Get the plan from Casdoor providing the plan_id. - - :param plan_id: the id of the plan - :return: a dict that contains plan's info - """ - url = self.endpoint + "/api/get-plan" - params = { - "id": f"{self.org_name}/{plan_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - plan = r.json() - return plan - - def modify_plan(self, method: str, plan: Plan) -> Dict: - url = self.endpoint + f"/api/{method}" - plan.owner = self.org_name - params = { - "id": f"{plan.owner}/{plan.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - plan_info = json.dumps(plan.to_dict()) - r = requests.post(url, params=params, data=plan_info) - response = r.json() - return response - - def add_plan(self, plan: Plan) -> Dict: - response = self.modify_plan("add-plan", plan) - return response - - def update_plan(self, plan: Plan) -> Dict: - response = self.modify_plan("update-plan", plan) - return response - - def delete_plan(self, plan: Plan) -> Dict: - response = self.modify_plan("delete-plan", plan) - return response - - def get_permissions(self) -> List[Dict]: - """ - Get the permissions from Casdoor. - - :return: a list of dicts containing permission info - """ - url = self.endpoint + "/api/get-permissions" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - permissions = r.json() - return permissions - - def get_permission(self, permission_id: str) -> Dict: - """ - Get the permission from Casdoor providing the permission_id. - - :param permission_id: the id of the permission - :return: a dict that contains permission's info - """ - url = self.endpoint + "/api/get-permission" - params = { - "id": f"{self.org_name}/{permission_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - permission = r.json() - return permission - - def modify_permission(self, method: str, permission: Permission) -> Dict: - url = self.endpoint + f"/api/{method}" - permission.owner = self.org_name - params = { - "id": f"{permission.owner}/{permission.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - permission_info = json.dumps(permission.to_dict()) - r = requests.post(url, params=params, data=permission_info) - response = r.json() - return response - - def add_permission(self, permission: Permission) -> Dict: - response = self.modify_permission("add-permission", permission) - return response - - def update_permission(self, permission: Permission) -> Dict: - response = self.modify_permission("update-permission", permission) - return response - - def delete_permission(self, permission: Permission) -> Dict: - response = self.modify_permission("delete-permission", permission) - return response - - def get_enforcers(self) -> List[Dict]: - """ - Get the enforcers from Casdoor. - - :return: a list of dicts containing enforcer info - """ - url = self.endpoint + "/api/get-enforcers" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - enforcers = r.json() - return enforcers - - def get_enforcer(self, enforcer_id: str) -> Dict: - """ - Get the enforcer from Casdoor providing the enforcer_id. - - :param enforcer_id: the id of the enforcer - :return: a dict that contains enforcer's info - """ - url = self.endpoint + "/api/get-enforcer" - params = { - "id": f"{self.org_name}/{enforcer_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - enforcer = r.json() - return enforcer - - def modify_enforcer(self, method: str, enforcer: Enforcer) -> Dict: - url = self.endpoint + f"/api/{method}" - enforcer.owner = self.org_name - params = { - "id": f"{enforcer.owner}/{enforcer.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - enforcer_info = json.dumps(enforcer.to_dict()) - r = requests.post(url, params=params, data=enforcer_info) - response = r.json() - return response - - def add_enforcer(self, enforcer: Enforcer) -> Dict: - response = self.modify_enforcer("add-enforcer", enforcer) - return response - - def update_enforcer(self, enforcer: Enforcer) -> Dict: - response = self.modify_enforcer("update-enforcer", enforcer) - return response - - def delete_enforcer(self, enforcer: Enforcer) -> Dict: - response = self.modify_enforcer("delete-enforcer", enforcer) - return response - - def get_resources(self, owner, user, field, value, sort_field, sort_order) -> List[Dict]: - """ - Get the resources from Casdoor. - - :return: a list of dicts containing resource info - """ - url = self.endpoint + "/api/get-resources" - params = { - "owner": owner, - "user": user, - "field": field, - "value": value, - "sortField": sort_field, - "sortOrder": sort_order, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - resources = r.json() - return resources - - def get_resource(self, resource_id: str) -> Dict: - """ - Get the resource from Casdoor providing the resource_id. - - :param resource_id: the id of the resource - :return: a dict that contains resource's info - """ - url = self.endpoint + "/api/get-resource" - params = { - "id": f"{self.org_name}/{resource_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - resource = r.json() - return resource - - def modify_resource(self, method: str, resource: Resource) -> Dict: - url = self.endpoint + f"/api/{method}" - resource.owner = self.org_name - params = { - "id": f"{resource.owner}/{resource.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - resource_info = json.dumps(resource.to_dict()) - r = requests.post(url, params=params, data=resource_info) - response = r.json() - return response - - def add_resource(self, resource: Resource) -> Dict: - response = self.modify_resource("add-resource", resource) - return response - - def update_resource(self, resource: Resource) -> Dict: - response = self.modify_resource("update-resource", resource) - return response - - def delete_resource(self, resource: Resource) -> Dict: - response = self.modify_resource("delete-resource", resource) - return response - - def get_tokens(self, p, page_size) -> List[Dict]: - """ - Get the tokens from Casdoor. - - :return: a list of dicts containing token info - """ - url = self.endpoint + "/api/get-tokens" - params = { - "owner": self.org_name, - "p": str(p), - "pageSize": str(page_size), - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - tokens = r.json() - return tokens - - def get_token(self, token_id: str) -> Dict: - """ - Get the token from Casdoor providing the token_id. - - :param token_id: the id of the token - :return: a dict that contains token's info - """ - url = self.endpoint + "/api/get-token" - params = { - "id": f"{self.org_name}/{token_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - token = r.json() - return token - - def modify_token(self, method: str, token: Token) -> Dict: - url = self.endpoint + f"/api/{method}" - token.owner = self.org_name - params = { - "id": f"{token.owner}/{token.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - token_info = json.dumps(token.to_dict()) - r = requests.post(url, params=params, data=token_info) - response = r.json() - return response - - def add_token(self, token: Token) -> Dict: - response = self.modify_token("add-token", token) - return response - - def update_token(self, token: Token) -> Dict: - response = self.modify_token("update-token", token) - return response - - def delete_token(self, token: Token) -> Dict: - response = self.modify_token("delete-token", token) - return response - - def get_sessions(self) -> List[Dict]: - """ - Get the sessions from Casdoor. - - :return: a list of dicts containing session info - """ - url = self.endpoint + "/api/get-sessions" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - sessions = r.json() - return sessions - - def get_session(self, session_id: str) -> Dict: - """ - Get the session from Casdoor providing the session_id. - - :param session_id: the id of the session - :return: a dict that contains session's info - """ - url = self.endpoint + "/api/get-session" - params = { - "id": f"{self.org_name}/{session_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - session = r.json() - return session - - def modify_session(self, method: str, session: Session) -> Dict: - url = self.endpoint + f"/api/{method}" - session.owner = self.org_name - params = { - "id": f"{session.owner}/{session.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - session_info = json.dumps(session.to_dict()) - r = requests.post(url, params=params, data=session_info) - response = r.json() - return response - - def add_session(self, session: Session) -> Dict: - response = self.modify_session("add-session", session) - return response - - def update_session(self, session: Session) -> Dict: - response = self.modify_session("update-session", session) - return response - - def delete_session(self, session: Session) -> Dict: - response = self.modify_session("delete-session", session) - return response - - def get_syncers(self) -> List[Dict]: - """ - Get the syncers from Casdoor. - - :return: a list of dicts containing syncer info - """ - url = self.endpoint + "/api/get-syncers" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - syncers = r.json() - return syncers - - def get_syncer(self, syncer_id: str) -> Dict: - """ - Get the syncer from Casdoor providing the syncer_id. - - :param syncer_id: the id of the syncer - :return: a dict that contains syncer's info - """ - url = self.endpoint + "/api/get-syncer" - params = { - "id": f"{self.org_name}/{syncer_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - syncer = r.json() - return syncer - - def modify_syncer(self, method: str, syncer: Syncer) -> Dict: - url = self.endpoint + f"/api/{method}" - syncer.owner = self.org_name - params = { - "id": f"{syncer.owner}/{syncer.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - syncer_info = json.dumps(syncer.to_dict()) - r = requests.post(url, params=params, data=syncer_info) - response = r.json() - return response - - def add_syncer(self, syncer: Syncer) -> Dict: - response = self.modify_syncer("add-syncer", syncer) - return response - - def update_syncer(self, syncer: Syncer) -> Dict: - response = self.modify_syncer("update-syncer", syncer) - return response - - def delete_syncer(self, syncer: Syncer) -> Dict: - response = self.modify_syncer("delete-syncer", syncer) - return response - - def get_webhooks(self) -> List[Dict]: - """ - Get the webhooks from Casdoor. - - :return: a list of dicts containing webhook info - """ - url = self.endpoint + "/api/get-webhooks" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - webhooks = r.json() - return webhooks - - def get_webhook(self, webhook_id: str) -> Dict: - """ - Get the webhook from Casdoor providing the webhook_id. - - :param webhook_id: the id of the webhook - :return: a dict that contains webhook's info - """ - url = self.endpoint + "/api/get-webhook" - params = { - "id": f"{self.org_name}/{webhook_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - webhook = r.json() - return webhook - - def modify_webhook(self, method: str, webhook: Webhook) -> Dict: - url = self.endpoint + f"/api/{method}" - webhook.owner = self.org_name - params = { - "id": f"{webhook.owner}/{webhook.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - webhook_info = json.dumps(webhook.to_dict()) - r = requests.post(url, params=params, data=webhook_info) - response = r.json() - return response - - def add_webhook(self, webhook: Webhook) -> Dict: - response = self.modify_webhook("add-webhook", webhook) - return response - - def update_webhook(self, webhook: Webhook) -> Dict: - response = self.modify_webhook("update-webhook", webhook) - return response - - def delete_webhook(self, webhook: Webhook) -> Dict: - response = self.modify_webhook("delete-webhook", webhook) - return response - - def get_subscriptions(self) -> List[Dict]: - """ - Get the subscriptions from Casdoor. - - :return: a list of dicts containing subscription info - """ - url = self.endpoint + "/api/get-subscriptions" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - subscriptions = r.json() - return subscriptions - - def get_subscription(self, subscription_id: str) -> Dict: - """ - Get the subscription from Casdoor providing the subscription_id. - - :param subscription_id: the id of the subscription - :return: a dict that contains subscription's info - """ - url = self.endpoint + "/api/get-subscription" - params = { - "id": f"{self.org_name}/{subscription_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - subscription = r.json() - return subscription - - def modify_subscription(self, method: str, subscription: Subscription) -> Dict: - url = self.endpoint + f"/api/{method}" - subscription.owner = self.org_name - params = { - "id": f"{subscription.owner}/{subscription.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - subscription_info = json.dumps(subscription.to_dict()) - r = requests.post(url, params=params, data=subscription_info) - response = r.json() - return response - - def add_subscription(self, subscription: Subscription) -> Dict: - response = self.modify_subscription("add-subscription", subscription) - return response - - def update_subscription(self, subscription: Subscription) -> Dict: - response = self.modify_subscription("update-subscription", subscription) - return response - - def delete_subscription(self, subscription: Subscription) -> Dict: - response = self.modify_subscription("delete-subscription", subscription) - return response - - def get_pricings(self) -> List[Dict]: - """ - Get the pricings from Casdoor. - - :return: a list of dicts containing pricing info - """ - url = self.endpoint + "/api/get-pricings" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - pricings = r.json() - return pricings - - def get_pricing(self, pricing_id: str) -> Dict: - """ - Get the pricing from Casdoor providing the pricing_id. - - :param pricing_id: the id of the pricing - :return: a dict that contains pricing's info - """ - url = self.endpoint + "/api/get-pricing" - params = { - "id": f"{self.org_name}/{pricing_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - pricing = r.json() - return pricing - - def modify_pricing(self, method: str, pricing: Pricing) -> Dict: - url = self.endpoint + f"/api/{method}" - pricing.owner = self.org_name - params = { - "id": f"{pricing.owner}/{pricing.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - pricing_info = json.dumps(pricing.to_dict()) - r = requests.post(url, params=params, data=pricing_info) - response = r.json() - return response - - def add_pricing(self, pricing: Pricing) -> Dict: - response = self.modify_pricing("add-pricing", pricing) - return response - - def update_pricing(self, pricing: Pricing) -> Dict: - response = self.modify_pricing("update-pricing", pricing) - return response - - def delete_pricing(self, pricing: Pricing) -> Dict: - response = self.modify_pricing("delete-pricing", pricing) - return response - - def get_products(self) -> List[Dict]: - """ - Get the products from Casdoor. - - :return: a list of dicts containing product info - """ - url = self.endpoint + "/api/get-products" - params = { - "owner": self.org_name, - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - products = r.json() - return products - - def get_product(self, product_id: str) -> Dict: - """ - Get the product from Casdoor providing the product_id. - - :param product_id: the id of the product - :return: a dict that contains product's info - """ - url = self.endpoint + "/api/get-product" - params = { - "id": f"{self.org_name}/{product_id}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - r = requests.get(url, params) - product = r.json() - return product - - def modify_product(self, method: str, product: Product) -> Dict: - url = self.endpoint + f"/api/{method}" - product.owner = self.org_name - params = { - "id": f"{product.owner}/{product.name}", - "clientId": self.client_id, - "clientSecret": self.client_secret, - } - product_info = json.dumps(product.to_dict()) - r = requests.post(url, params=params, data=product_info) - response = r.json() - return response - - def add_product(self, product: Product) -> Dict: - response = self.modify_product("add-product", product) - return response - - def update_product(self, product: Product) -> Dict: - response = self.modify_product("update-product", product) - return response - - def delete_product(self, product: Product) -> Dict: - response = self.modify_product("delete-product", product) - return response diff --git a/src/casdoor/model.py b/src/casdoor/model.py index 5606176..ed97f28 100644 --- a/src/casdoor/model.py +++ b/src/casdoor/model.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK from .user import User @@ -38,3 +44,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class ModelSDK(CasdoorSDK): + def get_models(self) -> List[Dict]: + """ + Get the models from Casdoor. + + :return: a list of dicts containing model info + """ + url = self.endpoint + "/api/get-models" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + models = r.json() + return models + + def get_model(self, model_id: str) -> Dict: + """ + Get the model from Casdoor providing the model_id. + + :param model_id: the id of the model + :return: a dict that contains model's info + """ + url = self.endpoint + "/api/get-model" + params = { + "id": f"{self.org_name}/{model_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + model = r.json() + return model + + def modify_model(self, method: str, model: Model) -> Dict: + url = self.endpoint + f"/api/{method}" + model.owner = self.org_name + params = { + "id": f"{model.owner}/{model.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + model_info = json.dumps(model.to_dict()) + r = requests.post(url, params=params, data=model_info) + response = r.json() + return response + + def add_model(self, model: Model) -> Dict: + response = self.modify_model("add-model", model) + return response + + def update_model(self, model: Model) -> Dict: + response = self.modify_model("update-model", model) + return response + + def delete_model(self, model: Model) -> Dict: + response = self.modify_model("delete-model", model) + return response diff --git a/src/casdoor/organization.py b/src/casdoor/organization.py index 5c854d7..c963610 100644 --- a/src/casdoor/organization.py +++ b/src/casdoor/organization.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class AccountItem: @@ -83,3 +89,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class OrganizationSDK(CasdoorSDK): + def get_organizations(self) -> List[Dict]: + """ + Get the organizations from Casdoor. + + :return: a list of dicts containing organization info + """ + url = self.endpoint + "/api/get-organizations" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + organizations = r.json() + return organizations + + def get_organization(self, organization_id: str) -> Dict: + """ + Get the organization from Casdoor providing the organization_id. + + :param organization_id: the id of the organization + :return: a dict that contains organization's info + """ + url = self.endpoint + "/api/get-organization" + params = { + "id": f"{self.org_name}/{organization_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + organization = r.json() + return organization + + def modify_organization(self, method: str, organization: Organization) -> Dict: + url = self.endpoint + f"/api/{method}" + organization.owner = self.org_name + params = { + "id": f"{organization.owner}/{organization.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + organization_info = json.dumps(organization.to_dict()) + r = requests.post(url, params=params, data=organization_info) + response = r.json() + return response + + def add_organization(self, organization: Organization) -> Dict: + response = self.modify_organization("add-organization", organization) + return response + + def update_organization(self, organization: Organization) -> Dict: + response = self.modify_organization("update-organization", organization) + return response + + def delete_organization(self, organization: Organization) -> Dict: + response = self.modify_organization("delete-organization", organization) + return response diff --git a/src/casdoor/payment.py b/src/casdoor/payment.py index 7692859..52833af 100644 --- a/src/casdoor/payment.py +++ b/src/casdoor/payment.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Payment: @@ -48,3 +54,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class PaymentSDK(CasdoorSDK): + def get_payments(self) -> List[Dict]: + """ + Get the payments from Casdoor. + + :return: a list of dicts containing payment info + """ + url = self.endpoint + "/api/get-payments" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + payments = r.json() + return payments + + def get_payment(self, payment_id: str) -> Dict: + """ + Get the payment from Casdoor providing the payment_id. + + :param payment_id: the id of the payment + :return: a dict that contains payment's info + """ + url = self.endpoint + "/api/get-payment" + params = { + "id": f"{self.org_name}/{payment_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + payment = r.json() + return payment + + def modify_payment(self, method: str, payment: Payment) -> Dict: + url = self.endpoint + f"/api/{method}" + payment.owner = self.org_name + params = { + "id": f"{payment.owner}/{payment.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + payment_info = json.dumps(payment.to_dict()) + r = requests.post(url, params=params, data=payment_info) + response = r.json() + return response + + def add_payment(self, payment: Payment) -> Dict: + response = self.modify_payment("add-payment", payment) + return response + + def update_payment(self, payment: Payment) -> Dict: + response = self.modify_payment("update-payment", payment) + return response + + def delete_payment(self, payment: Payment) -> Dict: + response = self.modify_payment("delete-payment", payment) + return response diff --git a/src/casdoor/permisssion.py b/src/casdoor/permisssion.py index 089a46e..e7dc54f 100644 --- a/src/casdoor/permisssion.py +++ b/src/casdoor/permisssion.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Permission: @@ -40,3 +46,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class PermissionSDK(CasdoorSDK): + def get_permissions(self) -> List[Dict]: + """ + Get the permissions from Casdoor. + + :return: a list of dicts containing permission info + """ + url = self.endpoint + "/api/get-permissions" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + permissions = r.json() + return permissions + + def get_permission(self, permission_id: str) -> Dict: + """ + Get the permission from Casdoor providing the permission_id. + + :param permission_id: the id of the permission + :return: a dict that contains permission's info + """ + url = self.endpoint + "/api/get-permission" + params = { + "id": f"{self.org_name}/{permission_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + permission = r.json() + return permission + + def modify_permission(self, method: str, permission: Permission) -> Dict: + url = self.endpoint + f"/api/{method}" + permission.owner = self.org_name + params = { + "id": f"{permission.owner}/{permission.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + permission_info = json.dumps(permission.to_dict()) + r = requests.post(url, params=params, data=permission_info) + response = r.json() + return response + + def add_permission(self, permission: Permission) -> Dict: + response = self.modify_permission("add-permission", permission) + return response + + def update_permission(self, permission: Permission) -> Dict: + response = self.modify_permission("update-permission", permission) + return response + + def delete_permission(self, permission: Permission) -> Dict: + response = self.modify_permission("delete-permission", permission) + return response diff --git a/src/casdoor/plan.py b/src/casdoor/plan.py index 36efbd4..6a1b962 100644 --- a/src/casdoor/plan.py +++ b/src/casdoor/plan.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Plan: @@ -32,3 +38,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class PlanSDK(CasdoorSDK): + def get_plans(self) -> List[Dict]: + """ + Get the plans from Casdoor. + + :return: a list of dicts containing plan info + """ + url = self.endpoint + "/api/get-plans" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + plans = r.json() + return plans + + def get_plan(self, plan_id: str) -> Dict: + """ + Get the plan from Casdoor providing the plan_id. + + :param plan_id: the id of the plan + :return: a dict that contains plan's info + """ + url = self.endpoint + "/api/get-plan" + params = { + "id": f"{self.org_name}/{plan_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + plan = r.json() + return plan + + def modify_plan(self, method: str, plan: Plan) -> Dict: + url = self.endpoint + f"/api/{method}" + plan.owner = self.org_name + params = { + "id": f"{plan.owner}/{plan.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + plan_info = json.dumps(plan.to_dict()) + r = requests.post(url, params=params, data=plan_info) + response = r.json() + return response + + def add_plan(self, plan: Plan) -> Dict: + response = self.modify_plan("add-plan", plan) + return response + + def update_plan(self, plan: Plan) -> Dict: + response = self.modify_plan("update-plan", plan) + return response + + def delete_plan(self, plan: Plan) -> Dict: + response = self.modify_plan("delete-plan", plan) + return response diff --git a/src/casdoor/pricing.py b/src/casdoor/pricing.py index fbbf583..5e3c695 100644 --- a/src/casdoor/pricing.py +++ b/src/casdoor/pricing.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Pricing: @@ -34,3 +40,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class PricingSDK(CasdoorSDK): + def get_pricings(self) -> List[Dict]: + """ + Get the pricings from Casdoor. + + :return: a list of dicts containing pricing info + """ + url = self.endpoint + "/api/get-pricings" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + pricings = r.json() + return pricings + + def get_pricing(self, pricing_id: str) -> Dict: + """ + Get the pricing from Casdoor providing the pricing_id. + + :param pricing_id: the id of the pricing + :return: a dict that contains pricing's info + """ + url = self.endpoint + "/api/get-pricing" + params = { + "id": f"{self.org_name}/{pricing_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + pricing = r.json() + return pricing + + def modify_pricing(self, method: str, pricing: Pricing) -> Dict: + url = self.endpoint + f"/api/{method}" + pricing.owner = self.org_name + params = { + "id": f"{pricing.owner}/{pricing.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + pricing_info = json.dumps(pricing.to_dict()) + r = requests.post(url, params=params, data=pricing_info) + response = r.json() + return response + + def add_pricing(self, pricing: Pricing) -> Dict: + response = self.modify_pricing("add-pricing", pricing) + return response + + def update_pricing(self, pricing: Pricing) -> Dict: + response = self.modify_pricing("update-pricing", pricing) + return response + + def delete_pricing(self, pricing: Pricing) -> Dict: + response = self.modify_pricing("delete-pricing", pricing) + return response diff --git a/src/casdoor/product.py b/src/casdoor/product.py index caf5632..6b7442b 100644 --- a/src/casdoor/product.py +++ b/src/casdoor/product.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK from .provider import Provider @@ -39,3 +45,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class ProductSDK(CasdoorSDK): + def get_products(self) -> List[Dict]: + """ + Get the products from Casdoor. + + :return: a list of dicts containing product info + """ + url = self.endpoint + "/api/get-products" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + products = r.json() + return products + + def get_product(self, product_id: str) -> Dict: + """ + Get the product from Casdoor providing the product_id. + + :param product_id: the id of the product + :return: a dict that contains product's info + """ + url = self.endpoint + "/api/get-product" + params = { + "id": f"{self.org_name}/{product_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + product = r.json() + return product + + def modify_product(self, method: str, product: Product) -> Dict: + url = self.endpoint + f"/api/{method}" + product.owner = self.org_name + params = { + "id": f"{product.owner}/{product.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + product_info = json.dumps(product.to_dict()) + r = requests.post(url, params=params, data=product_info) + response = r.json() + return response + + def add_product(self, product: Product) -> Dict: + response = self.modify_product("add-product", product) + return response + + def update_product(self, product: Product) -> Dict: + response = self.modify_product("update-product", product) + return response + + def delete_product(self, product: Product) -> Dict: + response = self.modify_product("delete-product", product) + return response diff --git a/src/casdoor/provider.py b/src/casdoor/provider.py index 7474bab..6dc7c8a 100644 --- a/src/casdoor/provider.py +++ b/src/casdoor/provider.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Provider: @@ -60,3 +66,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class ProviderSDK(CasdoorSDK): + def get_providers(self) -> List[Dict]: + """ + Get the providers from Casdoor. + + :return: a list of dicts containing provider info + """ + url = self.endpoint + "/api/get-providers" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + providers = r.json() + return providers + + def get_provider(self, provider_id: str) -> Dict: + """ + Get the provider from Casdoor providing the provider_id. + + :param provider_id: the id of the provider + :return: a dict that contains provider's info + """ + url = self.endpoint + "/api/get-provider" + params = { + "id": f"{self.org_name}/{provider_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + provider = r.json() + return provider + + def modify_provider(self, method: str, provider: Provider) -> Dict: + url = self.endpoint + f"/api/{method}" + provider.owner = self.org_name + params = { + "id": f"{provider.owner}/{provider.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + provider_info = json.dumps(provider.to_dict()) + r = requests.post(url, params=params, data=provider_info) + response = r.json() + return response + + def add_provider(self, provider: Provider) -> Dict: + response = self.modify_provider("add-provider", provider) + return response + + def update_provider(self, provider: Provider) -> Dict: + response = self.modify_provider("update-provider", provider) + return response + + def delete_provider(self, provider: Provider) -> Dict: + response = self.modify_provider("delete-provider", provider) + return response diff --git a/src/casdoor/resource.py b/src/casdoor/resource.py index 0ffdb02..c4e8b61 100644 --- a/src/casdoor/resource.py +++ b/src/casdoor/resource.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Resource: @@ -35,3 +41,68 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class ResourceSDK(CasdoorSDK): + def get_resources(self, owner, user, field, value, sort_field, sort_order) -> List[Dict]: + """ + Get the resources from Casdoor. + + :return: a list of dicts containing resource info + """ + url = self.endpoint + "/api/get-resources" + params = { + "owner": owner, + "user": user, + "field": field, + "value": value, + "sortField": sort_field, + "sortOrder": sort_order, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + resources = r.json() + return resources + + def get_resource(self, resource_id: str) -> Dict: + """ + Get the resource from Casdoor providing the resource_id. + + :param resource_id: the id of the resource + :return: a dict that contains resource's info + """ + url = self.endpoint + "/api/get-resource" + params = { + "id": f"{self.org_name}/{resource_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + resource = r.json() + return resource + + def modify_resource(self, method: str, resource: Resource) -> Dict: + url = self.endpoint + f"/api/{method}" + resource.owner = self.org_name + params = { + "id": f"{resource.owner}/{resource.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + resource_info = json.dumps(resource.to_dict()) + r = requests.post(url, params=params, data=resource_info) + response = r.json() + return response + + def add_resource(self, resource: Resource) -> Dict: + response = self.modify_resource("add-resource", resource) + return response + + def update_resource(self, resource: Resource) -> Dict: + response = self.modify_resource("update-resource", resource) + return response + + def delete_resource(self, resource: Resource) -> Dict: + response = self.modify_resource("delete-resource", resource) + return response diff --git a/src/casdoor/role.py b/src/casdoor/role.py index 1bef968..d70d401 100644 --- a/src/casdoor/role.py +++ b/src/casdoor/role.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Role: @@ -30,3 +36,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class RoleSDK(CasdoorSDK): + def get_roles(self) -> List[Dict]: + """ + Get the roles from Casdoor. + + :return: a list of dicts containing role info + """ + url = self.endpoint + "/api/get-roles" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + roles = r.json() + return roles + + def get_role(self, role_id: str) -> Dict: + """ + Get the role from Casdoor providing the role_id. + + :param role_id: the id of the role + :return: a dict that contains role's info + """ + url = self.endpoint + "/api/get-role" + params = { + "id": f"{self.org_name}/{role_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + role = r.json() + return role + + def modify_role(self, method: str, role: Role) -> Dict: + url = self.endpoint + f"/api/{method}" + role.owner = self.org_name + params = { + "id": f"{role.owner}/{role.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + role_info = json.dumps(role.to_dict()) + r = requests.post(url, params=params, data=role_info) + response = r.json() + return response + + def add_role(self, role: Role) -> Dict: + response = self.modify_role("add-role", role) + return response + + def update_role(self, role: Role) -> Dict: + response = self.modify_role("update-role", role) + return response + + def delete_role(self, role: Role) -> Dict: + response = self.modify_role("delete-role", role) + return response diff --git a/src/casdoor/session.py b/src/casdoor/session.py index 7207f83..1d27d57 100644 --- a/src/casdoor/session.py +++ b/src/casdoor/session.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Session: @@ -26,3 +32,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class SessionSDK(CasdoorSDK): + def get_sessions(self) -> List[Dict]: + """ + Get the sessions from Casdoor. + + :return: a list of dicts containing session info + """ + url = self.endpoint + "/api/get-sessions" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + sessions = r.json() + return sessions + + def get_session(self, session_id: str) -> Dict: + """ + Get the session from Casdoor providing the session_id. + + :param session_id: the id of the session + :return: a dict that contains session's info + """ + url = self.endpoint + "/api/get-session" + params = { + "id": f"{self.org_name}/{session_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + session = r.json() + return session + + def modify_session(self, method: str, session: Session) -> Dict: + url = self.endpoint + f"/api/{method}" + session.owner = self.org_name + params = { + "id": f"{session.owner}/{session.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + session_info = json.dumps(session.to_dict()) + r = requests.post(url, params=params, data=session_info) + response = r.json() + return response + + def add_session(self, session: Session) -> Dict: + response = self.modify_session("add-session", session) + return response + + def update_session(self, session: Session) -> Dict: + response = self.modify_session("update-session", session) + return response + + def delete_session(self, session: Session) -> Dict: + response = self.modify_session("delete-session", session) + return response diff --git a/src/casdoor/subscription.py b/src/casdoor/subscription.py index 045bdc6..9efbb62 100644 --- a/src/casdoor/subscription.py +++ b/src/casdoor/subscription.py @@ -11,8 +11,13 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. - +import json from datetime import datetime +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Subscription: @@ -38,3 +43,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class SubscriptionSDK(CasdoorSDK): + def get_subscriptions(self) -> List[Dict]: + """ + Get the subscriptions from Casdoor. + + :return: a list of dicts containing subscription info + """ + url = self.endpoint + "/api/get-subscriptions" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + subscriptions = r.json() + return subscriptions + + def get_subscription(self, subscription_id: str) -> Dict: + """ + Get the subscription from Casdoor providing the subscription_id. + + :param subscription_id: the id of the subscription + :return: a dict that contains subscription's info + """ + url = self.endpoint + "/api/get-subscription" + params = { + "id": f"{self.org_name}/{subscription_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + subscription = r.json() + return subscription + + def modify_subscription(self, method: str, subscription: Subscription) -> Dict: + url = self.endpoint + f"/api/{method}" + subscription.owner = self.org_name + params = { + "id": f"{subscription.owner}/{subscription.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + subscription_info = json.dumps(subscription.to_dict()) + r = requests.post(url, params=params, data=subscription_info) + response = r.json() + return response + + def add_subscription(self, subscription: Subscription) -> Dict: + response = self.modify_subscription("add-subscription", subscription) + return response + + def update_subscription(self, subscription: Subscription) -> Dict: + response = self.modify_subscription("update-subscription", subscription) + return response + + def delete_subscription(self, subscription: Subscription) -> Dict: + response = self.modify_subscription("delete-subscription", subscription) + return response diff --git a/src/casdoor/syncer.py b/src/casdoor/syncer.py index c4e9d64..65b00a1 100644 --- a/src/casdoor/syncer.py +++ b/src/casdoor/syncer.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class TableColumn: @@ -57,3 +63,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class SyncerSDK(CasdoorSDK): + def get_syncers(self) -> List[Dict]: + """ + Get the syncers from Casdoor. + + :return: a list of dicts containing syncer info + """ + url = self.endpoint + "/api/get-syncers" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + syncers = r.json() + return syncers + + def get_syncer(self, syncer_id: str) -> Dict: + """ + Get the syncer from Casdoor providing the syncer_id. + + :param syncer_id: the id of the syncer + :return: a dict that contains syncer's info + """ + url = self.endpoint + "/api/get-syncer" + params = { + "id": f"{self.org_name}/{syncer_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + syncer = r.json() + return syncer + + def modify_syncer(self, method: str, syncer: Syncer) -> Dict: + url = self.endpoint + f"/api/{method}" + syncer.owner = self.org_name + params = { + "id": f"{syncer.owner}/{syncer.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + syncer_info = json.dumps(syncer.to_dict()) + r = requests.post(url, params=params, data=syncer_info) + response = r.json() + return response + + def add_syncer(self, syncer: Syncer) -> Dict: + response = self.modify_syncer("add-syncer", syncer) + return response + + def update_syncer(self, syncer: Syncer) -> Dict: + response = self.modify_syncer("update-syncer", syncer) + return response + + def delete_syncer(self, syncer: Syncer) -> Dict: + response = self.modify_syncer("delete-syncer", syncer) + return response diff --git a/src/casdoor/token.py b/src/casdoor/token.py index 18e459d..4341b88 100644 --- a/src/casdoor/token.py +++ b/src/casdoor/token.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK class Token: @@ -36,3 +42,65 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class TokenSDK(CasdoorSDK): + def get_tokens(self, p, page_size) -> List[Dict]: + """ + Get the tokens from Casdoor. + + :return: a list of dicts containing token info + """ + url = self.endpoint + "/api/get-tokens" + params = { + "owner": self.org_name, + "p": str(p), + "pageSize": str(page_size), + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + tokens = r.json() + return tokens + + def get_token(self, token_id: str) -> Dict: + """ + Get the token from Casdoor providing the token_id. + + :param token_id: the id of the token + :return: a dict that contains token's info + """ + url = self.endpoint + "/api/get-token" + params = { + "id": f"{self.org_name}/{token_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + token = r.json() + return token + + def modify_token(self, method: str, token: Token) -> Dict: + url = self.endpoint + f"/api/{method}" + token.owner = self.org_name + params = { + "id": f"{token.owner}/{token.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + token_info = json.dumps(token.to_dict()) + r = requests.post(url, params=params, data=token_info) + response = r.json() + return response + + def add_token(self, token: Token) -> Dict: + response = self.modify_token("add-token", token) + return response + + def update_token(self, token: Token) -> Dict: + response = self.modify_token("update-token", token) + return response + + def delete_token(self, token: Token) -> Dict: + response = self.modify_token("delete-token", token) + return response diff --git a/src/casdoor/user.py b/src/casdoor/user.py index 22cf189..4601df8 100644 --- a/src/casdoor/user.py +++ b/src/casdoor/user.py @@ -12,6 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK + class User: def __init__(self): @@ -51,3 +58,86 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class UserSDK(CasdoorSDK): + def get_users(self) -> List[Dict]: + """ + Get the users from Casdoor. + + :return: a list of dicts containing user info + """ + url = self.endpoint + "/api/get-users" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + users = r.json() + return users + + def get_user(self, user_id: str) -> Dict: + """ + Get the user from Casdoor providing the user_id. + + :param user_id: the id of the user + :return: a dict that contains user's info + """ + url = self.endpoint + "/api/get-user" + params = { + "id": f"{self.org_name}/{user_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + user = r.json() + return user + + def get_user_count(self, is_online: bool = None) -> int: + """ + Get the count of filtered users for an organization + :param is_online: True for online users, False for offline users, + None for all users + :return: the count of filtered users for an organization + """ + url = self.endpoint + "/api/get-user-count" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + + if is_online is None: + params["isOnline"] = "" + else: + params["isOnline"] = "1" if is_online else "0" + + r = requests.get(url, params) + count = r.json() + return count + + def modify_user(self, method: str, user: User) -> Dict: + url = self.endpoint + f"/api/{method}" + user.owner = self.org_name + params = { + "id": f"{user.owner}/{user.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + user_info = json.dumps(user.to_dict()) + r = requests.post(url, params=params, data=user_info) + response = r.json() + return response + + def add_user(self, user: User) -> Dict: + response = self.modify_user("add-user", user) + return response + + def update_user(self, user: User) -> Dict: + response = self.modify_user("update-user", user) + return response + + def delete_user(self, user: User) -> Dict: + response = self.modify_user("delete-user", user) + return response diff --git a/src/casdoor/webhook.py b/src/casdoor/webhook.py index 0a2e8c2..b8eb008 100644 --- a/src/casdoor/webhook.py +++ b/src/casdoor/webhook.py @@ -11,6 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import json +from typing import Dict, List + +import requests + +from src.casdoor import CasdoorSDK from .syncer import TableColumn @@ -43,3 +49,63 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + + +class WebhookSDK(CasdoorSDK): + def get_webhooks(self) -> List[Dict]: + """ + Get the webhooks from Casdoor. + + :return: a list of dicts containing webhook info + """ + url = self.endpoint + "/api/get-webhooks" + params = { + "owner": self.org_name, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + webhooks = r.json() + return webhooks + + def get_webhook(self, webhook_id: str) -> Dict: + """ + Get the webhook from Casdoor providing the webhook_id. + + :param webhook_id: the id of the webhook + :return: a dict that contains webhook's info + """ + url = self.endpoint + "/api/get-webhook" + params = { + "id": f"{self.org_name}/{webhook_id}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + webhook = r.json() + return webhook + + def modify_webhook(self, method: str, webhook: Webhook) -> Dict: + url = self.endpoint + f"/api/{method}" + webhook.owner = self.org_name + params = { + "id": f"{webhook.owner}/{webhook.name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + webhook_info = json.dumps(webhook.to_dict()) + r = requests.post(url, params=params, data=webhook_info) + response = r.json() + return response + + def add_webhook(self, webhook: Webhook) -> Dict: + response = self.modify_webhook("add-webhook", webhook) + return response + + def update_webhook(self, webhook: Webhook) -> Dict: + response = self.modify_webhook("update-webhook", webhook) + return response + + def delete_webhook(self, webhook: Webhook) -> Dict: + response = self.modify_webhook("delete-webhook", webhook) + return response