cc

gitea/apiobject.py

@@ -5,6 +5,7 @@ from typing import List, Tuple, Dict, Sequence, Optional, Union, Set
 from .baseapiobject import ReadonlyApiObject, ApiObject
 from .exceptions import *
 
+
 class Organization(ApiObject):
     """see https://try.gitea.io/api/swagger#/organization/orgGetAll"""
 
@@ -233,7 +234,7 @@ class Branch(ReadonlyApiObject):
 
     _fields_to_parsers = {
         # This is not a commit object
-        #"commit": lambda gitea, c: Commit.parse_response(gitea, c)
+        # "commit": lambda gitea, c: Commit.parse_response(gitea, c)
     }
 
     @classmethod
@@ -269,7 +270,7 @@ class Repository(ApiObject):
     _fields_to_parsers = {
         # dont know how to tell apart user and org as owner except form email being empty.
         "owner": lambda gitea, r: Organization.parse_response(gitea, r)
-                                if r["email"] == "" else User.parse_response(gitea, r),
+        if r["email"] == "" else User.parse_response(gitea, r),
         "updated_at": lambda gitea, t: Util.convert_time(t),
     }
 
@@ -374,9 +375,9 @@ class Repository(ApiObject):
         )
         return Issue.parse_response(self.gitea, result)
 
-    def create_milestone(self, title: str, description: str, due_date: str = None, state:str = "open") -> "Milestone":
+    def create_milestone(self, title: str, description: str, due_date: str = None, state: str = "open") -> "Milestone":
         url = Repository.REPO_MILESTONES.format(owner=self.owner.username, repo=self.name)
-        data= {"title": title, "description": description, "state": state}
+        data = {"title": title, "description": description, "state": state}
         if due_date: data["due_date"] = due_date
         result = self.gitea.requests_post(url, data=data)
         return Milestone.parse_response(self.gitea, result)
@@ -440,14 +441,14 @@ class Repository(ApiObject):
         self.gitea.requests_post(url, data=data)
         # TODO: make sure this instance is either updated or discarded
 
-    def get_git_content(self: str = None, commit : "Commit" = None) -> List["Content"]:
+    def get_git_content(self: str = None, commit: "Commit" = None) -> List["Content"]:
         """https://git.sopranium.de/api/swagger#/repository/repoGetContentsList"""
         url = Repository.REPO_CONTENTS.format(owner=self.owner.username, repo=self.name)
         data = {"ref": "HEAD" if commit is None else commit.sha}
         result = [Content.parse_response(self.gitea, f) for f in self.gitea.requests_get(url, data)]
         return result
 
-    def get_file_content(self, content: "Content", commit : "Commit" = None) -> Union[str, List["Content"]]:
+    def get_file_content(self, content: "Content", commit: "Commit" = None) -> Union[str, List["Content"]]:
         """https://git.sopranium.de/api/swagger#/repository/repoGetContents"""
         url = Repository.REPO_CONTENT.format(owner=self.owner.username,
                                              repo=self.name, filepath=content.path)
@@ -459,10 +460,11 @@ class Repository(ApiObject):
 
     def delete(self):
         self.gitea.requests_delete(
-                Repository.REPO_DELETE % (self.owner.username, self.name)
+            Repository.REPO_DELETE % (self.owner.username, self.name)
         )
         self.deleted = True
 
+
 class Milestone(ApiObject):
     API_OBJECT = """/repos/{owner}/{repo}/milestones/{number}"""  # <owner, repo>
 
@@ -701,6 +703,7 @@ class Team(ApiObject):
         url = f"/teams/{self.id}/members/{user_name}"
         self.gitea.requests_delete(url)
 
+
 class Content(ReadonlyApiObject):
     FILE = "file"
 
@@ -723,4 +726,3 @@ class Util:
             return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S%z")
         except ValueError:
             return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S")
-

gitea/baseapiobject.py

@@ -1,4 +1,5 @@
-from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation,RawRequestEndpointMissing
+from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation, RawRequestEndpointMissing
+
 
 class ReadonlyApiObject:
 
@@ -53,7 +54,7 @@ class ReadonlyApiObject:
             cls._add_read_property(name, value, api_object)
         # add all patchable fields missing in the request to be writable
         for name in cls._fields_to_parsers.keys():
-            if not hasattr(api_object,name):
+            if not hasattr(api_object, name):
                 cls._add_read_property(name, None, api_object)
 
     @classmethod
@@ -73,7 +74,6 @@ class ReadonlyApiObject:
 
 
 class ApiObject(ReadonlyApiObject):
-
     _patchable_fields = set()
 
     def __init__(self, gitea):
@@ -97,13 +97,13 @@ class ApiObject(ReadonlyApiObject):
 
     @classmethod
     def _initialize(cls, gitea, api_object, result):
-        super()._initialize(gitea,api_object,result)
+        super()._initialize(gitea, api_object, result)
         for name, value in result.items():
             if name in cls._patchable_fields:
-                cls._add_write_property(name,value,api_object)
+                cls._add_write_property(name, value, api_object)
         # add all patchable fields missing in the request to be writable
         for name in cls._patchable_fields:
-            if not hasattr(api_object,name):
+            if not hasattr(api_object, name):
                 cls._add_write_property(name, None, api_object)
 
     @classmethod

gitea/exceptions.py

@@ -13,11 +13,13 @@ class ObjectIsInvalid(Exception):
 class ConflictException(Exception):
     pass
 
+
 class RawRequestEndpointMissing(Exception):
     """This ApiObject can only be obtained through other api objects and does not have
     diret .request method."""
     pass
 
+
 class MissiongEqualyImplementation(Exception):
     """
     Each Object obtained from the gitea api must be able to check itself for equality in relation to its

gitea/gitea.py

@@ -6,8 +6,9 @@ import requests
 from frozendict import frozendict
 from httpcache import CachingHTTPAdapter
 
-from .exceptions import NotFoundException, ConflictException, AlreadyExistsException
 from .apiobject import User, Organization, Repository, Team
+from .exceptions import NotFoundException, ConflictException, AlreadyExistsException
+
 
 class Gitea:
     """ Object to establish a session with Gitea. """
@@ -45,7 +46,7 @@ class Gitea:
             return json.loads(result.text)
         return {}
 
-    def requests_get(self, endpoint:str , params= frozendict(), sudo=None):
+    def requests_get(self, endpoint: str, params=frozendict(), sudo=None):
         combined_params = {}
         combined_params.update(params)
         if sudo:
@@ -64,7 +65,7 @@ class Gitea:
             raise Exception(message)
         return self.parse_result(request)
 
-    def requests_get_paginated(self, endpoint:str , params = frozendict(), sudo=None, page_key: str = "page"):
+    def requests_get_paginated(self, endpoint: str, params=frozendict(), sudo=None, page_key: str = "page"):
         page = 1
         combined_params = {}
         combined_params.update(params)
@@ -91,19 +92,19 @@ class Gitea:
             self.logger.error(message)
             raise Exception(message)
 
-    def requests_post(self, endpoint:str , data:dict):
+    def requests_post(self, endpoint: str, data: dict):
         request = self.requests.post(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data))
         if request.status_code not in [200, 201, 202]:
             if ("already exists" in request.text or "e-mail already in use" in request.text):
                 self.logger.warning(request.text)
                 raise AlreadyExistsException()
-            self.logger.error(f"Received status code: {request.status_code} ({ request.url})")
+            self.logger.error(f"Received status code: {request.status_code} ({request.url})")
             self.logger.error(f"With info: {data} ({self.headers})")
             self.logger.error(f"Answer: {request.text}")
             raise Exception(f"Received status code: {request.status_code} ({request.url}), {request.text}")
         return self.parse_result(request)
 
-    def requests_patch(self, endpoint:str, data:dict):
+    def requests_patch(self, endpoint: str, data: dict):
         request = self.requests.patch(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data))
         if request.status_code not in [200, 201]:
             error_message = f"Received status code: {request.status_code} ({request.url}) {data}"
@@ -147,14 +148,14 @@ class Gitea:
         return None
 
     def create_user(
-        self,
+            self,
-        user_name: str,
+            user_name: str,
-        email: str,
+            email: str,
-        password: str,
+            password: str,
-        login_name: str = None,
+            login_name: str = None,
-        change_pw=True,
+            change_pw=True,
-        send_notify=True,
+            send_notify=True,
-        source_id=0,
+            source_id=0,
     ):
         """ Create User.
         Throws:
@@ -190,16 +191,16 @@ class Gitea:
         return user
 
     def create_repo(
-        self,
+            self,
-        repoOwner: Union[User,Organization],
+            repoOwner: Union[User, Organization],
-        repoName: str,
+            repoName: str,
-        description: str = "",
+            description: str = "",
-        private: bool = False,
+            private: bool = False,
-        autoInit=True,
+            autoInit=True,
-        gitignores: str = None,
+            gitignores: str = None,
-        license: str = None,
+            license: str = None,
-        readme: str = "Default",
+            readme: str = "Default",
-        issue_labels: str = None,
+            issue_labels: str = None,
     ):
         """ Create a Repository.
         Throws:
@@ -231,13 +232,13 @@ class Gitea:
         return Repository.parse_response(self, result)
 
     def create_org(
-        self,
+            self,
-        owner: User,
+            owner: User,
-        orgName: str,
+            orgName: str,
-        description: str,
+            description: str,
-        location="",
+            location="",
-        website="",
+            website="",
-        full_name="",
+            full_name="",
     ):
         assert isinstance(owner, User)
         result = self.requests_post(
@@ -265,20 +266,20 @@ class Gitea:
         return Organization.parse_response(self, result)
 
     def create_team(
-        self,
+            self,
-        org: Organization,
+            org: Organization,
-        name: str,
+            name: str,
-        description: str = "",
+            description: str = "",
-        permission: str = "read",
+            permission: str = "read",
-        units=(
+            units=(
-            "repo.code",
+                    "repo.code",
-            "repo.issues",
+                    "repo.issues",
-            "repo.ext_issues",
+                    "repo.ext_issues",
-            "repo.wiki",
+                    "repo.wiki",
-            "repo.pulls",
+                    "repo.pulls",
-            "repo.releases",
+                    "repo.releases",
-            "repo.ext_wiki",
+                    "repo.ext_wiki",
-        ),
+            ),
     ):
         """ Creates a Team.