Skip to content

Commit

Permalink
🚧 Refactoring of the GogsService & adaptation on new features.
Browse files Browse the repository at this point in the history 
- extraction of all GogsApi code into an adapter subclass
- removal of custom features to use the RepositoryServices equivalents

Fixes: #18
Signed-off-by: Guyzmo <[email protected]>
  • Loading branch information
@guyzmo
guyzmo committed Feb 2, 2017
1 parent b0a8694 commit 5c37245
Show file tree
Hide file tree
Showing 2 changed files with 90 additions and 88 deletions.
176 changes: 89 additions & 87 deletions git_repo/services/ext/gogs.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,51 +6,94 @@
from ..service import register_target, RepositoryService, os
from ...exceptions import ResourceError, ResourceExistsError, ResourceNotFoundError

import gogs_client
import requests
from gogs_client import GogsApi, GogsRepo, Token, UsernamePassword, ApiFailure
from requests import Session, HTTPError
from urllib.parse import urlparse, urlunparse
import functools

from git import config as git_config
from git.exc import GitCommandError

class GogsClient(GogsApi):
def __init__(self, *args, **kwarg):
self.session = Session()
super().__init__(*args, session=self.session, **kwarg)

def set_token(self, token):
self.auth = Token(token)

def set_default_private(self, p):
self.default_private = p

def setup_session(self, ssl_config, proxy=dict()):
self.session.verify = ssl_config
self.session.proxies.update(proxy)

@property
def username(self):
if not hasattr(self, '_username'):
self._username = self.authenticated_user(self.auth).username
return self._username

def orgs(self):
orgs = self._check_ok(self._get('/user/orgs', auth=self.auth)).json()
#return [gogs_client.GogsUser.from_json(org) for org in orgs]
return [org['username'] for org in orgs]

def create_repository(self, user, repo):
if user == self.username:
repository = self.create_repo(self.auth, name=repo, private=self.default_private)
elif user in self.orgs():
data = dict(name=repo, private=self.default_private)
response = self._post('/org/{}/repos'.format(user), auth=self.auth, data=data)
repository = GogsRepo.from_json(self._check_ok(response).json())
else:
data = dict(name=repo, private=self.default_private)
response = self._post('/admin/users/{}/repos'.format(user), auth=self.auth, data=data)
repository = GogsRepo.from_json(self._check_ok(response).json())

def delete_repository(self, user, repo):
return self.delete_repo(self.auth, user, repo)

def repository(self, user, repo):
return self.get_repo(self.auth, user, repo)

def repositories(self, user):
r = self._get('/user/repos', auth=self.auth)
repositories = self._check_ok(r).json()
repositories = [repo for repo in repositories if repo['owner']['username'] == user]
return repositories

@register_target('gg', 'gogs')
class GogsService(RepositoryService):
fqdn = 'try.gogs.io'
#fqdn = 'http://127.0.0.1:3000'
gg = None

def __init__(self, *args, **kwargs):
self.session = requests.Session()
RepositoryService.__init__(self, *args, **kwargs)
self.ensure_init()

def ensure_init(self):
if self.gg is not None:
return
self.url_base, self.fqdn = self._url_parse(self.fqdn)
if 'insecure' not in self.config:
self.insecure = self.fqdn != 'try.gogs.io'
self.session.verify = not self.insecure
if 'server-cert' in self.config:
self.session.verify = self.config['server-cert']
self.default_private = self.config.get('default-private', 'false').lower() not in ('0','no','false')
self.ssh_url = self.config.get('ssh-url', None) or self.fqdn
if not self.repository:
config = git_config.GitConfigParser(os.path.join(os.environ['HOME'], '.gitconfig'), True)
else:
config = self.repository.config_reader()
proxies = {}
for scheme in 'http https'.split():
proxy = config.get_value(scheme, 'proxy', '')
if proxy:
proxies[scheme] = proxy
self.session.proxies.update(proxies)
self.gg = gogs_client.GogsApi(self.url_base, self.session)
#if ':' in self._privatekey:
# self.auth = gogs_client.UsernamePassword(*self._privatekey.split(':',1))
#else:
self.auth = gogs_client.Token(self._privatekey)
self.gg = GogsClient(self.url_base)

super().__init__(*args, **kwargs)

self.gg.set_token(self._privatekey)
self.gg.set_default_private(self.default_create_private)
self.gg.setup_session(
self.session_certificate or not self.session_insecure,
self.session_proxy)

def connect(self):
try:
self.username = self.user # Call to self.gg.authenticated_user()
except HTTPError as err:
if err.response is not None and err.response.status_code == 401:
if not self._privatekey:
raise ConnectionError('Could not connect to GoGS. '
'Please configure .gitconfig '
'with your gogs private key.') from err
else:
raise ConnectionError('Could not connect to GoGS. '
'Check your configuration and try again.') from err
else:
raise err

@classmethod
def _url_parse(cls, url):
Expand All @@ -75,70 +118,31 @@ def url_rw(self):
@classmethod
def get_auth_token(cls, login, password, prompt=None):
import platform
name = 'git-repo2 token used on {}'.format(platform.node())
name = 'git-repo token used on {}'.format(platform.node())
if '/' in login:
url, login = login.rsplit('/', 1)
else:
url = input('URL [{}]> '.format(cls.fqdn)) or cls.fqdn
url_base, fqdn = cls._url_parse(url)
gg = gogs_client.GogsApi(url_base)
auth = gogs_client.UsernamePassword(login, password)
gg = GogsApi(url_base)
auth = UsernamePassword(login, password)
tokens = gg.get_tokens(auth, login)
tokens = dict((token.name, token.token) for token in tokens)
if name in tokens:
return tokens[name]
if 'git-repo2 token' in tokens:
return tokens['git-repo2 token']
if 'git-repo token' in tokens:
return tokens['git-repo token']
token = gg.create_token(auth, name, login)
return token.token

@property
def user(self):
return self.gg.authenticated_user(self.auth).username

def orgs(self):
orgs = self.gg._check_ok(self.gg._get('/user/orgs', auth=self.auth)).json()
#return [gogs_client.GogsUser.from_json(org) for org in orgs]
return [org['username'] for org in orgs]

def connect(self):
self.ensure_init()
try:
if self.insecure:
try:
try:
urllib3 = requests.packages.urllib3
except Exception:
import urllib3
urllib3.disable_warnings()
except ImportError:
pass
self.username = self.user # Call to self.gg.authenticated_user()
except requests.HTTPError as err:
if err.response is not None and err.response.status_code == 401:
if not self._privatekey:
raise ConnectionError('Could not connect to GoGS. '
'Please configure .gitconfig '
'with your gogs private key.') from err
else:
raise ConnectionError('Could not connect to GoGS. '
'Check your configuration and try again.') from err
else:
raise err
return self.gg.username

def create(self, user, repo, add=False):
try:
if user == self.username:
repository = self.gg.create_repo(self.auth, name=repo, private=self.default_private)
elif user in self.orgs():
data = dict(name=repo, private=self.default_private)
response = self.gg._post('/org/{}/repos'.format(user), auth=self.auth, data=data)
repository = gogs_client.GogsRepo.from_json(self.gg._check_ok(response).json())
else:
data = dict(name=repo, private=self.default_private)
response = self.gg._post('/admin/users/{}/repos'.format(user), auth=self.auth, data=data)
repository = gogs_client.GogsRepo.from_json(self.gg._check_ok(response).json())
except gogs_client.ApiFailure as err:
self.gg.create_repository(user, repo)
except ApiFailure as err:
if err.status_code == 422:
raise ResourceExistsError("Project already exists.") from err
else:
Expand All @@ -155,8 +159,8 @@ def delete(self, repo, user=None):
if not user:
user = self.username
try:
self.gg.delete_repo(self.auth, user, repo)
except gogs_client.ApiFailure as err:
self.gg.delete_repository(user, repo)
except ApiFailure as err:
if err.status_code == 404:
raise ResourceNotFoundError("Cannot delete: repository {}/{} does not exists.".format(user, repo)) from err
elif err.status_code == 403:
Expand Down Expand Up @@ -190,9 +194,7 @@ def col_print(lines, indent=0, pad=2):
for row in rows:
print(" "*indent + (" "*pad).join(line.ljust(col_width) for line in row))

r = self.gg._get('/user/repos', auth=self.auth)
repositories = self.gg._check_ok(r).json()
repositories = [repo for repo in repositories if repo['owner']['username'] == user]
repositories = self.gg.repositories(user)
if user != self.username and not repositories and user not in self.orgs:
raise ResourceNotFoundError("Unable to list namespace {} - only authenticated user and orgs available for listing.".format(user))
if not _long:
Expand Down Expand Up @@ -227,8 +229,8 @@ def col_print(lines, indent=0, pad=2):

def get_repository(self, user, repo):
try:
return self.gg.get_repo(self.auth, user, repo)
except gogs_client.ApiFailure as err:
return self.gg.repository(user, repo)
except ApiFailure as err:
if err.status_code == 404:
raise ResourceNotFoundError("Cannot get: repository {}/{} does not exists.".format(user, repo)) from err
raise ResourceError("Unhandled error: {}".format(err)) from err
Expand Down
2 changes: 1 addition & 1 deletion tests/integration/test_gogs.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ def get_service(self):
# return gogs.GogsService(c=dict(fqdn=self.fqdn,__name__='gitrepo "gogs"',token=os.environ['PRIVATE_KEY_GOGS']))

def get_requests_session(self):
return self.service.session
return self.service.gg.session

def test_00_fork(self):
pass
Expand Down

0 comments on commit 5c37245

Please sign in to comment.