import requests from gogs_client._implementation.http_utils import RelativeHttpRequestor, append_url from gogs_client.auth import Token from gogs_client.entities import GogsUser, GogsRepo, GogsOrg, GogsTeam class GogsApi(object): """ A Gogs client, serving as a wrapper around the Gogs HTTP API. """ def __init__(self, base_url, session=None): """ :param str base_url: the URL of the Gogs server to communicate with. Should be given with the https protocol :param requests.Session session: a ``requests`` session instance """ api_base = append_url(base_url, "/api/v1/") self._requestor = RelativeHttpRequestor(api_base, session=session) def valid_authentication(self, auth): """ Returns whether ``auth`` is valid :param auth.Authentication auth: authentication object :return: whether the provided authentication is valid :rtype: bool :raises NetworkFailure: if there is an error communicating with the server """ return self._get("/user", auth=auth).ok def authenticated_user(self, auth): """ Returns the user authenticated by ``auth`` :param auth.Authentication auth: authentication for user to retrieve :return: user authenticated by the provided authentication :rtype: GogsUser :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ response = self._get("/user", auth=auth) return GogsUser.from_json(self._check_ok(response).json()) def get_tokens(self, auth, username=None): """ Returns the tokens owned by the specified user. If no user is specified, uses the user authenticated by ``auth``. :param auth.Authentication auth: authentication for user to retrieve. Must be a username-password authentication, due to a restriction of the Gogs API :param str username: username of owner of tokens :return: list of tokens :rtype: List[Token] :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ if username is None: username = self.authenticated_user(auth).username response = self._get("/users/{u}/tokens".format(u=username), auth=auth) return [Token.from_json(o) for o in self._check_ok(response).json()] def create_token(self, auth, name, username=None): """ Creates a new token with the specified name for the specified user. If no user is specified, uses user authenticated by ``auth``. :param auth.Authentication auth: authentication for user to retrieve. Must be a username-password authentication, due to a restriction of the Gogs API :param str name: name of new token :param str username: username of owner of new token :return: new token representation :rtype: Token :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ if username is None: username = self.authenticated_user(auth).username data = {"name": name} response = self._post("/users/{u}/tokens".format(u=username), auth=auth, data=data) return Token.from_json(self._check_ok(response).json()) def ensure_token(self, auth, name, username=None): """ Ensures the existence of a token with the specified name for the specified user. Creates a new token if none exists. If no user is specified, uses user authenticated by ``auth``. :param auth.Authentication auth: authentication for user to retrieve. Must be a username-password authentication, due to a restriction of the Gogs API :param str name: name of new token :param str username: username of owner of new token :return: token representation :rtype: Token :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ if username is None: username = self.authenticated_user(auth).username tokens = [token for token in self.get_tokens(auth, username) if token.name == name] if len(tokens) > 0: return tokens[0] return self.create_token(auth, name, username) def create_repo(self, auth, name, description=None, private=False, auto_init=False, gitignore_templates=None, license_template=None, readme_template=None, organization=None): """ Creates a new repository, and returns the created repository. :param auth.Authentication auth: authentication object :param str name: name of the repository to create :param str description: description of the repository to create :param bool private: whether the create repository should be private :param bool auto_init: whether the created repository should be auto-initialized with an initial commit :param list[str] gitignore_templates: collection of ``.gitignore`` templates to apply :param str license_template: license template to apply :param str readme_template: README template to apply :param str organization: organization under which repository is created :return: a representation of the created repository :rtype: GogsRepo :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ gitignores = None if gitignore_templates is None \ else ",".join(gitignore_templates) data = { "name": name, "description": description, "private": private, "auto_init": auto_init, "gitignores": gitignores, "license": license_template, "readme": readme_template } data = {k: v for (k, v) in data.items() if v is not None} url = "/org/{0}/repos".format(organization) if organization else "/user/repos" response = self._post(url, auth=auth, data=data) return GogsRepo.from_json(self._check_ok(response).json()) def repo_exists(self, auth, username, repo_name): """ Returns whether a repository with name ``repo_name`` owned by the user with username ``username`` exists. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :param str repo_name: name of repository :return: whether the repository exists :rtype: bool :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ path = "/repos/{u}/{r}".format(u=username, r=repo_name) return self._get(path, auth=auth).ok def get_repo(self, auth, username, repo_name): """ Returns a the repository with name ``repo_name`` owned by the user with username ``username``. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :param str repo_name: name of repository :return: a representation of the retrieved repository :rtype: GogsRepo :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ path = "/repos/{u}/{r}".format(u=username, r=repo_name) response = self._check_ok(self._get(path, auth=auth)) return GogsRepo.from_json(response.json()) def get_user_repos(self, auth, username): """ Returns the repositories owned by the user with username ``username``. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :return: a list of repositories :rtype: List[GogsRepo] :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ path = "/users/{u}/repos".format(u=username) response = self._check_ok(self._get(path, auth=auth)) return [GogsRepo.from_json(repo_json) for repo_json in response.json()] def delete_repo(self, auth, username, repo_name): """ Deletes the repository with name ``repo_name`` owned by the user with username ``username``. :param auth.Authentication auth: authentication object :param str username: username of owner of repository to delete :param str repo_name: name of repository to delete :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ path = "/repos/{u}/{r}".format(u=username, r=repo_name) self._check_ok(self._delete(path, auth=auth)) def migrate_repo(self, auth, clone_addr, uid, repo_name, auth_username=None, auth_password=None, mirror=False, private=False, description=None): """ Migrate a repository from another Git hosting source for the authenticated user. :param auth.Authentication auth: authentication object :param str clone_addr: Remote Git address (HTTP/HTTPS URL or local path) :param int uid: user ID of repository owner :param str repo_name: Repository name :param bool mirror: Repository will be a mirror. Default is false :param bool private: Repository will be private. Default is false :param str description: Repository description :return: a representation of the migrated repository :rtype: GogsRepo :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ # "auth_username": auth_username, # "auth_password": auth_password, data = { "clone_addr": clone_addr, "uid": uid, "repo_name": repo_name, "mirror": mirror, "private": private, "description": description, } data = {k: v for (k, v) in data.items() if v is not None} url = "/repos/migrate" response = self._post(url, auth=auth, data=data) return GogsRepo.from_json(self._check_ok(response).json()) def create_user(self, auth, login_name, username, email, password, send_notify=False): """ Creates a new user, and returns the created user. :param auth.Authentication auth: authentication object, must be admin-level :param str login_name: login name for created user :param str username: username for created user :param str email: email address for created user :param str password: password for created user :param bool send_notify: whether a notification email should be sent upon creation :return: a representation of the created user :rtype: GogsUser :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ # Since the source_id parameter was not well-documented at the time this method was # written, force the default value data = { "login_name": login_name, "username": username, "email": email, "password": password, "send_notify": send_notify } response = self._post("/admin/users", auth=auth, data=data) self._check_ok(response) return GogsUser.from_json(response.json()) def user_exists(self, username): """ Returns whether a user with username ``username`` exists. :param str username: username of user :return: whether a user with the specified username exists :rtype: bool :raises NetworkFailure: if there is an error communicating with the server :return: """ path = "/users/{}".format(username) return self._get(path).ok def search_users(self, username_keyword, limit=10): """ Searches for users whose username matches ``username_keyword``, and returns a list of matched users. :param str username_keyword: keyword to search with :param int limit: maximum number of returned users :return: a list of matched users :rtype: List[GogsUser] :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ params = {"q": username_keyword, "limit": limit} response = self._check_ok(self._get("/users/search", params=params)) return [GogsUser.from_json(user_json) for user_json in response.json()["data"]] def get_user(self, auth, username): """ Returns a representing the user with username ``username``. :param auth.Authentication auth: authentication object, can be ``None`` :param str username: username of user to get :return: the retrieved user :rtype: GogsUser :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ path = "/users/{}".format(username) response = self._check_ok(self._get(path, auth=auth)) return GogsUser.from_json(response.json()) def update_user(self, auth, username, update): """ Updates the user with username ``username`` according to ``update``. :param auth.Authentication auth: authentication object, must be admin-level :param str username: username of user to update :param GogsUserUpdate update: a ``GogsUserUpdate`` object describing the requested update :return: the updated user :rtype: GogsUser :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ path = "/admin/users/{}".format(username) response = self._check_ok(self._patch(path, auth=auth, data=update.as_dict())) return GogsUser.from_json(response.json()) def delete_user(self, auth, username): """ Deletes the user with username ``username``. Should only be called if the to-be-deleted user has no repositories. :param auth.Authentication auth: authentication object, must be admin-level :param str username: username of user to delete """ path = "/admin/users/{}".format(username) self._check_ok(self._delete(path, auth=auth)) def get_repo_hooks(self, auth, username, repo_name): """ Returns all hooks of repository with name ``repo_name`` owned by the user with username ``username``. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :param str repo_name: name of repository :return: a list of hooks for the specified repository :rtype: List[GogsRepo.Hooks] :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ path = "/repos/{u}/{r}/hooks".format(u=username, r=repo_name) response = self._check_ok(self._get(path, auth=auth)) return [GogsRepo.Hook.from_json(hook) for hook in response.json()] def create_hook(self, auth, repo_name, hook_type, config, events=None, organization=None, active=False): """ Creates a new hook, and returns the created hook. :param auth.Authentication auth: authentication object, must be admin-level :param str repo_name: the name of the repo for which we create the hook :param str hook_type: The type of webhook, either "gogs" or "slack" :param dict config: Settings for this hook (possible keys are ``"url"``, ``"content_type"``, ``"secret"``) :param list events: Determines what events the hook is triggered for. Default: ["push"] :param str organization: Organization of the repo :param bool active: Determines whether the hook is actually triggered on pushes. Default is false :return: a representation of the created hook :rtype: GogsRepo.Hook :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ if events is None: events = ["push"] # default value is mutable, so assign inside body data = { "type": hook_type, "config": config, "events": events, "active": active } url = "/repos/{o}/{r}/hooks".format(o=organization, r=repo_name) if organization is not None \ else "/repos/{r}/hooks".format(r=repo_name) response = self._post(url, auth=auth, data=data) self._check_ok(response) return GogsRepo.Hook.from_json(response.json()) def update_hook(self, auth, repo_name, hook_id, update, organization=None): """ Updates hook with id ``hook_id`` according to ``update``. :param auth.Authentication auth: authentication object :param str repo_name: repo of the hook to update :param int hook_id: id of the hook to update :param GogsHookUpdate update: a ``GogsHookUpdate`` object describing the requested update :param str organization: name of associated organization, if applicable :return: the updated hook :rtype: GogsRepo.Hook :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ if organization is not None: path = "/repos/{o}/{r}/hooks/{i}".format(o=organization, r=repo_name, i=hook_id) else: path = "/repos/{r}/hooks/{i}".format(r=repo_name, i=hook_id) response = self._check_ok(self._patch(path, auth=auth, data=update.as_dict())) return GogsRepo.Hook.from_json(response.json()) def delete_hook(self, auth, username, repo_name, hook_id): """ Deletes the hook with id ``hook_id`` for repo with name ``repo_name`` owned by the user with username ``username``. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :param str repo_name: name of repository of hook to delete :param int hook_id: id of hook to delete :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ path = "/repos/{u}/{r}/hooks/{i}".format(u=username, r=repo_name, i=hook_id) self._check_ok(self._delete(path, auth=auth)) def create_organization(self, auth, owner_name, org_name, full_name=None, description=None, website=None, location=None): """ Creates a new organization, and returns the created organization. :param auth.Authentication auth: authentication object, must be admin-level :param str owner_name: Username of organization owner :param str org_name: Organization name :param str full_name: Full name of organization :param str description: Description of the organization :param str website: Official website :param str location: Organization location :return: a representation of the created organization :rtype: GogsOrg :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ data = { "username": org_name, "full_name": full_name, "description": description, "website": website, "location": location } url = "/admin/users/{u}/orgs".format(u=owner_name) response = self._post(url, auth=auth, data=data) self._check_ok(response) return GogsOrg.from_json(response.json()) def create_organization_team(self, auth, org_name, name, description=None, permission="read"): """ Creates a new team of the organization. :param auth.Authentication auth: authentication object, must be admin-level :param str org_name: Organization user name :param str name: Full name of the team :param str description: Description of the team :param str permission: Team permission, can be read, write or admin, default is read :return: a representation of the created team :rtype: GogsTeam :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ data = { "name": name, "description": description, "permission": permission } url = "/admin/orgs/{o}/teams".format(o=org_name) response = self._post(url, auth=auth, data=data) self._check_ok(response) return GogsTeam.from_json(response.json()) def add_team_membership(self, auth, team_id, username): """ Add user to team. :param auth.Authentication auth: authentication object, must be admin-level :param str team_id: Team's id :param str username: Username of the user to be added to team :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ url = "/admin/teams/{t}/members/{u}".format(t=team_id, u=username) self._check_ok(self._put(url, auth=auth)) def remove_team_membership(self, auth, team_id, username): """ Remove user from team. :param auth.Authentication auth: authentication object, must be admin-level :param str team_id: Team's id :param str username: Username of the user to be removed from the team :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ url = "/admin/teams/{t}/members/{u}".format(t=team_id, u=username) self._check_ok(self._delete(url, auth=auth)) def add_repo_to_team(self, auth, team_id, repo_name): """ Add or update repo from team. :param auth.Authentication auth: authentication object, must be admin-level :param str team_id: Team's id :param str repo_name: Name of the repo to be added to the team :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ url = "/admin/teams/{t}/repos/{r}".format(t=team_id, r=repo_name) self._check_ok(self._put(url, auth=auth)) def remove_repo_from_team(self, auth, team_id, repo_name): """ Remove repo from team. :param auth.Authentication auth: authentication object, must be admin-level :param str team_id: Team's id :param str repo_name: Name of the repo to be removed from the team :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ url = "/admin/teams/{t}/repos/{r}".format(t=team_id, r=repo_name) self._check_ok(self._delete(url, auth=auth)) def list_deploy_keys(self, auth, username, repo_name): """ List deploy keys for the specified repo. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :param str repo_name: the name of the repo :return: a list of deploy keys for the repo :rtype: List[GogsRepo.DeployKey] :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ response = self._check_ok(self._get("/repos/{u}/{r}/keys".format(u=username, r=repo_name), auth=auth)) return [GogsRepo.DeployKey.from_json(key_json) for key_json in response.json()] def get_deploy_key(self, auth, username, repo_name, key_id): """ Get a deploy key for the specified repo. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :param str repo_name: the name of the repo :param int key_id: the id of the key :return: the deploy key :rtype: GogsRepo.DeployKey :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ response = self._check_ok( self._get("/repos/{u}/{r}/keys/{k}".format(u=username, r=repo_name, k=key_id), auth=auth)) return GogsRepo.DeployKey.from_json(response.json()) def add_deploy_key(self, auth, username, repo_name, title, key_content): """ Add a deploy key to the specified repo. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :param str repo_name: the name of the repo :param str title: title of the key to add :param str key_content: content of the key to add :return: a representation of the added deploy key :rtype: GogsRepo.DeployKey :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ data = { "title": title, "key": key_content } response = self._check_ok( self._post("/repos/{u}/{r}/keys".format(u=username, r=repo_name), auth=auth, data=data)) return GogsRepo.DeployKey.from_json(response.json()) def delete_deploy_key(self, auth, username, repo_name, key_id): """ Remove deploy key for the specified repo. :param auth.Authentication auth: authentication object :param str username: username of owner of repository :param str repo_name: the name of the repo :param int key_id: the id of the key :raises NetworkFailure: if there is an error communicating with the server :raises ApiFailure: if the request cannot be serviced """ response = self._check_ok( self._delete("/repos/{u}/{r}/keys/{k}".format(u=username, r=repo_name, k=key_id), auth=auth)) self._check_ok(response) # Helper methods def _delete(self, path, auth=None, **kwargs): if auth is not None: auth.update_kwargs(kwargs) try: return self._requestor.delete(path, **kwargs) except requests.RequestException as exc: raise NetworkFailure(exc) def _get(self, path, auth=None, **kwargs): if auth is not None: auth.update_kwargs(kwargs) try: return self._requestor.get(path, **kwargs) except requests.RequestException as exc: raise NetworkFailure(exc) def _patch(self, path, auth=None, **kwargs): if auth is not None: auth.update_kwargs(kwargs) try: return self._requestor.patch(path, **kwargs) except requests.RequestException as exc: raise NetworkFailure(exc) def _post(self, path, auth=None, **kwargs): if auth is not None: auth.update_kwargs(kwargs) try: return self._requestor.post(path, **kwargs) except requests.RequestException as exc: raise NetworkFailure(exc) def _put(self, path, auth=None, **kwargs): if auth is not None: auth.update_kwargs(kwargs) try: return self._requestor.put(path, **kwargs) except requests.RequestException as exc: raise NetworkFailure(exc) @staticmethod def _check_ok(response): """ Raise exception if response is non-OK, otherwise return response """ if not response.ok: GogsApi._fail(response) return response @staticmethod def _fail(response): """ Raise an ApiFailure pertaining to the given response """ message = "Status code: {}-{}, url: {}".format(response.status_code, response.reason, response.url) try: message += ", message:{}".format(response.json()["message"]) except (ValueError, KeyError): pass raise ApiFailure(message, response.status_code) class ApiFailure(Exception): """ Raised to signal a failed request """ def __init__(self, message, status_code): self._message = message self._status_code = status_code def __str__(self): return self._message @property def message(self): """ An error message explaining why the request failed. :type: str """ return self._message @property def status_code(self): """ The HTTP status code of the response to the failed request :type: int """ return self._status_code class NetworkFailure(Exception): """ Raised to signal a network-level failure """ def __init__(self, cause=None): self._cause = cause def __str__(self): return "Caused by: {}".format(str(self._cause)) @property def cause(self): """ The exception causing the network failure :type: Exception """ return self._cause