kopia lustrzana https://github.com/Langenfeld/py-gitea
split things into sepreate files
Langenfeld
rodzic
0888bc0af4
commit
ac38e81b5e
|
|
@ -0,0 +1,70 @@
|
|||
import logging
|
||||
from .exceptions import ObjectIsInvalid
|
||||
|
||||
class BasicGiteaApiObject:
|
||||
|
||||
GET_API_OBJECT = "FORMAT/STINING/{argument}"
|
||||
PATCH_API_OBJECT = "FORMAT/STINING/{argument}"
|
||||
|
||||
def __init__(self, gitea, id):
|
||||
self.__id = id
|
||||
self.gitea = gitea
|
||||
self.deleted = False # set if .delete was called, so that an exception is risen
|
||||
self.dirty_fields = set()
|
||||
|
||||
def __eq__(self, other):
|
||||
return other.id == self.id if isinstance(other, type(self)) else False
|
||||
|
||||
def __str__(self):
|
||||
return "GiteaAPIObject (%s) id: %s"%(type(self),self.id)
|
||||
|
||||
def __hash__(self):
|
||||
return self.id
|
||||
|
||||
fields_to_parsers = {}
|
||||
|
||||
def commit(self):
|
||||
raise NotImplemented()
|
||||
|
||||
def get_dirty_fields(self):
|
||||
return {name: getattr(self,name) for name in self.dirty_fields}
|
||||
|
||||
@classmethod
|
||||
def parse_request(cls, gitea, result):
|
||||
id = int(result["id"])
|
||||
logging.debug("Found api object of type %s (id: %s)" % (type(cls), id))
|
||||
api_object = cls(gitea, id=id)
|
||||
cls._initialize(gitea, api_object, result)
|
||||
return api_object
|
||||
|
||||
@classmethod
|
||||
def _get_gitea_api_object(cls, gitea, args):
|
||||
"""Retrieving an object always as GET_API_OBJECT """
|
||||
return gitea.requests_get(cls.GET_API_OBJECT.format(**args))
|
||||
|
||||
patchable_fields = set()
|
||||
|
||||
@classmethod
|
||||
def _initialize(cls, gitea, api_object, result):
|
||||
for name, value in result.items():
|
||||
if name in cls.fields_to_parsers and value is not None:
|
||||
parse_func = cls.fields_to_parsers[name]
|
||||
value = parse_func(gitea, value)
|
||||
if name in cls.patchable_fields:
|
||||
prop = property(
|
||||
(lambda name: lambda self: self.__get_var(name))(name),
|
||||
(lambda name: lambda self, v: self.__set_var(name, v))(name))
|
||||
else:
|
||||
prop = property(
|
||||
(lambda name: lambda self: self.__get_var(name))(name))
|
||||
setattr(cls, name, prop)
|
||||
setattr(api_object, "_"+name, value)
|
||||
|
||||
def __set_var(self,name,i):
|
||||
self.dirty_fields.add(name)
|
||||
setattr(self,"_"+name,i)
|
||||
|
||||
def __get_var(self,name):
|
||||
if self.deleted:
|
||||
raise ObjectIsInvalid()
|
||||
return getattr(self,"_"+name)
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
class AlreadyExistsException(Exception):
|
||||
pass
|
||||
|
||||
class NotFoundException(Exception):
|
||||
pass
|
||||
|
||||
class ObjectIsInvalid(Exception):
|
||||
pass
|
||||
111
gitea/gitea.py
111
gitea/gitea.py
|
|
@ -3,103 +3,13 @@ import json
|
|||
import requests
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from httpcache import CachingHTTPAdapter
|
||||
from .giteaApiObject import GiteaApiObject
|
||||
from .basicGiteaApiObject import BasicGiteaApiObject
|
||||
from .exceptions import *
|
||||
|
||||
logging = logging.getLogger("gitea")
|
||||
|
||||
class AlreadyExistsException(Exception):
|
||||
pass
|
||||
|
||||
class NotFoundException(Exception):
|
||||
pass
|
||||
|
||||
class ObjectIsInvalid(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class GiteaApiObject:
|
||||
|
||||
GET_API_OBJECT = "FORMAT/STINING/{argument}"
|
||||
PATCH_API_OBJECT = "FORMAT/STINING/{argument}"
|
||||
|
||||
def __init__(self, gitea, id):
|
||||
self.__id = id
|
||||
self.gitea = gitea
|
||||
self.deleted = False # set if .delete was called, so that an exception is risen
|
||||
self.dirty_fields = set()
|
||||
|
||||
def __eq__(self, other):
|
||||
return other.id == self.id if isinstance(other, type(self)) else False
|
||||
|
||||
def __str__(self):
|
||||
return "GiteaAPIObject (%s) id: %s"%(type(self),self.id)
|
||||
|
||||
def __hash__(self):
|
||||
return self.id
|
||||
|
||||
fields_to_parsers = {}
|
||||
|
||||
def commit(self):
|
||||
raise NotImplemented()
|
||||
|
||||
def get_dirty_fields(self):
|
||||
return {name: getattr(self,name) for name in self.dirty_fields}
|
||||
|
||||
@classmethod
|
||||
def request(cls, gitea, id):
|
||||
"""Use for ginving a nice e.g. 'request(gita, orgname, repo, ticket)'.
|
||||
All args are put into an args tuple for passing around"""
|
||||
return cls._request(gitea, {"id": id})
|
||||
|
||||
@classmethod
|
||||
def _request(cls, gitea, args):
|
||||
result = cls._get_gitea_api_object(gitea, args)
|
||||
api_object = cls.parse_request(gitea, result)
|
||||
for key, value in args.items(): # hack: not all necessary request args in api result (e.g. repo name in issue)
|
||||
if not hasattr(api_object, key):
|
||||
setattr(api_object, key, value)
|
||||
return api_object
|
||||
|
||||
@classmethod
|
||||
def parse_request(cls, gitea, result):
|
||||
id = int(result["id"])
|
||||
logging.debug("Found api object of type %s (id: %s)" % (type(cls), id))
|
||||
api_object = cls(gitea, id=id)
|
||||
cls._initialize(gitea, api_object, result)
|
||||
return api_object
|
||||
|
||||
@classmethod
|
||||
def _get_gitea_api_object(cls, gitea, args):
|
||||
"""Retrieving an object always as GET_API_OBJECT """
|
||||
return gitea.requests_get(cls.GET_API_OBJECT.format(**args))
|
||||
|
||||
patchable_fields = set()
|
||||
|
||||
@classmethod
|
||||
def _initialize(cls, gitea, api_object, result):
|
||||
for name, value in result.items():
|
||||
if name in cls.fields_to_parsers and value is not None:
|
||||
parse_func = cls.fields_to_parsers[name]
|
||||
value = parse_func(gitea, value)
|
||||
if name in cls.patchable_fields:
|
||||
prop = property(
|
||||
(lambda name: lambda self: self.__get_var(name))(name),
|
||||
(lambda name: lambda self, v: self.__set_var(name, v))(name))
|
||||
else:
|
||||
prop = property(
|
||||
(lambda name: lambda self: self.__get_var(name))(name))
|
||||
setattr(cls, name, prop)
|
||||
setattr(api_object, "_"+name, value)
|
||||
|
||||
def __set_var(self,name,i):
|
||||
self.dirty_fields.add(name)
|
||||
setattr(self,"_"+name,i)
|
||||
|
||||
def __get_var(self,name):
|
||||
if self.deleted:
|
||||
raise ObjectIsInvalid()
|
||||
return getattr(self,"_"+name)
|
||||
|
||||
|
||||
class Organization(GiteaApiObject):
|
||||
|
||||
GET_API_OBJECT = """/orgs/{name}""" # <org>
|
||||
|
|
@ -216,7 +126,7 @@ class User(GiteaApiObject):
|
|||
@classmethod
|
||||
def request(cls, gitea, name):
|
||||
api_object = cls._request(gitea, {"name": name})
|
||||
api_object.update_mail()
|
||||
#api_object.update_mail()
|
||||
return api_object
|
||||
|
||||
patchable_fields = {"active", "admin", "allow_create_organization", "allow_git_hook", "allow_import_local",
|
||||
|
|
@ -400,7 +310,7 @@ class Milestone(GiteaApiObject):
|
|||
return str(vars(self))
|
||||
|
||||
|
||||
class Comment(GiteaApiObject):
|
||||
class Comment(BasicGiteaApiObject):
|
||||
|
||||
GET_API_OBJECT = """NONE"""
|
||||
PATCH_API_OBJECT = "/repos/{owner}/{repo}/issues/comments/{id}"
|
||||
|
|
@ -416,11 +326,6 @@ class Comment(GiteaApiObject):
|
|||
|
||||
patchable_fields = {"body"}
|
||||
|
||||
@classmethod
|
||||
def request(cls, gitea, args):
|
||||
raise NotImplemented("this method is not supported by the Gitea Api")
|
||||
|
||||
|
||||
class Issue(GiteaApiObject):
|
||||
|
||||
GET_API_OBJECT = """/repos/{owner}/{repo}/issues/{number}""" # <owner, repo, index>
|
||||
|
|
@ -603,7 +508,9 @@ class Gitea:
|
|||
"Content-type": "application/json",
|
||||
}
|
||||
self.url = url
|
||||
self.requests = requests
|
||||
self.requests = requests.Session()
|
||||
self.requests.mount('http://', CachingHTTPAdapter())
|
||||
self.requests.mount('https://', CachingHTTPAdapter())
|
||||
|
||||
def get_url(self, endpoint):
|
||||
""" Returns the full API-URL.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,25 @@
|
|||
from .basicGiteaApiObject import BasicGiteaApiObject
|
||||
import logging
|
||||
|
||||
class GiteaApiObject(BasicGiteaApiObject):
|
||||
|
||||
GET_API_OBJECT = "FORMAT/STINING/{argument}"
|
||||
PATCH_API_OBJECT = "FORMAT/STINING/{argument}"
|
||||
|
||||
def __init__(self, gitea, id):
|
||||
super(GiteaApiObject, self).__init__(gitea, id)
|
||||
|
||||
@classmethod
|
||||
def request(cls, gitea, id):
|
||||
"""Use for ginving a nice e.g. 'request(gita, orgname, repo, ticket)'.
|
||||
All args are put into an args tuple for passing around"""
|
||||
return cls._request(gitea, {"id": id})
|
||||
|
||||
@classmethod
|
||||
def _request(cls, gitea, args):
|
||||
result = cls._get_gitea_api_object(gitea, args)
|
||||
api_object = cls.parse_request(gitea, result)
|
||||
for key, value in args.items(): # hack: not all necessary request args in api result (e.g. repo name in issue)
|
||||
if not hasattr(api_object, key):
|
||||
setattr(api_object, key, value)
|
||||
return api_object
|
||||
|
|
@ -3,10 +3,10 @@ import os
|
|||
import pytest
|
||||
import uuid
|
||||
|
||||
from .gitea import Gitea, User, Organization, Team, Repository, version
|
||||
from .gitea import Gitea, User, Organization, Team, Repository
|
||||
from .gitea import NotFoundException, AlreadyExistsException
|
||||
|
||||
assert version >= "0.4.0"
|
||||
|
||||
gitea = None
|
||||
|
||||
# put a ".token" file into your directory containg only the token for gitea
|
||||
|
|
|
|||
Ładowanie…
Reference in New Issue