Initial commit

Signed-off-by: René Jochum <rene@jochums.at>
master
René Jochum 6 years ago
commit 32a020a22b
No known key found for this signature in database
GPG Key ID: 9E8B1C32F5F318A9

8
.gitignore vendored

@ -0,0 +1,8 @@
.vscode/
venv/
repositories/
vcs-mirrors.yaml
__pycache__/
*.egg-info

@ -0,0 +1,7 @@
Changelog
=========
This document describes changes between each past release.
0.0.1 (unreleased)
------------------

@ -0,0 +1,5 @@
Contributors
============
* Sam Gleske - Idea and some code from https://github.com/samrocketman
* René Jochum <rene@jochums.at>

@ -0,0 +1,19 @@
Copyright (c) 2018 René Jochum
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.

@ -0,0 +1,81 @@
vcs-mirrors
===========
A python-only clone of https://github.com/samrocketman/gitlab-mirrors/ with a YAML config file.
Requirements
------------
- Python 3.5+ (Debian Stretch+, Ubuntu Xenial+)
- virtualenv if you don't want to mess with System Python
- git-remote-bzr https://github.com/felipec/git-remote-bzr for Bazaar support
Features
--------
* Mirror different types of source repositories: Bazaar, Git, Subversion. Mirror all into git.
* GitLab mirror adding.
* When adding a mirror if the project doesn't exist in GitLab it will be auto-created.
* Set project creation defaults (e.g. issues enabled, wiki enabled, etc.)
* Github mirror adding.
* Same as with Gitlab.
* mirror anything to Git (not just Gitlab and Github).
* Update a single mirror.
* Update all known mirrors.
Installation
++++++++++++
On Debian
---------
For Bazaar support:
$ apt install git-remote-bzr
Install into a virtualenv:
$ virtualenv -p /usr/bin/python3 --no-site-packages venv
$ venv/bin/pip install "vcs-mirrors[gitlab,github]"
Then copy vcs-mirrors.yaml.example into your current-working-directory:
$ cp venv/lib/python3.6/site-packages/vcs-mirrors/vcs-mirrors.yaml.sample .
Edit it for your needs.
Usage
+++++
venv/bin/vcs-mirrors -h
venv/bin/vcs-mirrors add -h
add examples:
-------------
This one try to create a repo "pcdummy/proxmox-dockerfiles" on git.lxch.eu - the identifier must be unique in the config file:
$ vcs-mirrors add me/p-dockerfiles https://github.com/pcdummy/proxmox-dockerfiles.git git.lxch.eu:pcdummy/proxmox-dockerfiles
This doesn't:
$ vcs-mirrors add me/p-dockerfiles https://github.com/pcdummy/proxmox-dockerfiles.git git@git.lxch.eu:pcdummy/proxmox-dockerfiles.git
Full mirroring include "prune" and "force" pull/push:
$ vcs-mirrors add -f -p me/p-dockerfiles https://github.com/pcdummy/proxmox-dockerfiles.git git.lxch.eu:pcdummy/proxmox-dockerfiles
If you give an host as target "add" creates the repo on the host and translates it to a git URL else add does nothing else than adding the params to your configuration file.
Development
+++++++++++
pip install -e ."[development,gitlab,github]"
Keywords
++++++++
gitlab github sync mirror vcs-mirror

@ -0,0 +1,76 @@
import codecs
import os
from setuptools import setup, find_packages
# abspath here because setup.py may be __main__, in which case
# __file__ is not guaranteed to be absolute
here = os.path.abspath(os.path.dirname(__file__))
def read_file(filename):
"""Open a related file and return its content."""
with codecs.open(os.path.join(here, filename), encoding='utf-8') as f:
content = f.read()
return content
README = read_file('README.rst')
CHANGELOG = read_file('CHANGELOG.rst')
CONTRIBUTORS = read_file('CONTRIBUTORS.rst')
REQUIREMENTS = [
'ruamel.yaml',
'zope.interface',
]
GITLAB_REQUIRES = [
'python-gitlab',
]
GITHUB_REQUIRES = [
'pygithub',
]
DEVELOPMENT_REQUIRES = [
'pylint',
'autopep8',
'flake8',
'ipython',
]
DEPENDENCY_LINKS = []
ENTRY_POINTS = {
'console_scripts': [
'vcs-mirrors = vcs_mirrors.scripts.main:main'
]
}
setup(name='vcs_mirrors',
version='0.0.1.dev0',
description='',
long_description='{}\n\n{}\n\n{}'.format(README, CHANGELOG, CONTRIBUTORS),
license='MIT',
classifiers=[
'Programming Language :: Python',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'License :: OSI Approved :: MIT'
],
keywords='Console VCS mirror vcs-mirrors',
author='René Jochum',
author_email='rene@jochums.at',
url='https://github.com/pcdummy/vcs_mirrors',
packages=find_packages(),
package_data={'': ['*.rst', '*.py', '*.yaml']},
include_package_data=True,
zip_safe=False,
install_requires=REQUIREMENTS,
extras_require={
'gitlab': GITLAB_REQUIRES,
'github': GITHUB_REQUIRES,
'development': DEVELOPMENT_REQUIRES,
},
dependency_links=DEPENDENCY_LINKS,
entry_points=ENTRY_POINTS)

@ -0,0 +1,42 @@
settings:
# Absolute or relative path to the store of repositories
local_path: repositories/
# Hosts
hosts:
gitlab.com:
type: gitlab
url: gitlab.com # Optional
api_key: <your-api-key> # Replace with your key
ssl_verify: true # Default: True
public: true # Default: False
issues_enabled: true # Default: False
wall_enabled: true # Default: False
merge_requests_enabled: true # Default: False
wiki_enabled: true # Default: False
snippets_enabled: true # Default: False
use_https: false # Defualt: False
github.com:
type: github
api_key: <your-api-key> # Replace with your key
public: true # Default: True
issues_enabled: true # Default: True
wiki_enabled: true # Default: True
downloads_enabled: true # Default: True
projects_enabled: true # Default: True
use_https: false # Default: False
git.lxch.eu:
type: gitlab
api_key: <your-api-key> # Replace with your key
public: true # Default: False
issues_enabled: true # Default: False
wall_enabled: true # Default: False
merge_requests_enabled: true # Default: False
wiki_enabled: true # Default: False
snippets_enabled: true # Default: False
use_https: false # Defualt: False
# Repos
repos:

@ -0,0 +1,96 @@
import logging
from zope.interface.verify import verifyObject
from zope.interface.exceptions import BrokenImplementation
from vcs_mirrors.lib.interfaces import IHost
from vcs_mirrors.lib.loader import load_hosts
from vcs_mirrors.lib.utils import get_url_host
def configure_argparse(subparser):
subparser.add_argument('-f', '--force',
help='force fetch and push (default False)',
dest='tags',
action='store_true',
default=False,
required=False)
subparser.add_argument('-p', '--prune',
help='Prune on fetch and push (default False)',
dest='tags',
action='store_true',
default=False,
required=False)
subparser.add_argument('name',
help='Internal repository name, for example pcdummy/vcs-mirrors')
subparser.add_argument('source',
help='The source VCS repo URL')
subparser.add_argument('dest',
help='The destination VCS repo URL or <HOST>:<REPO>, if <REPO> is not given we use "name" as REPO')
def execute(config, args):
if args['name'] in config['repos']:
logging.fatal('The repository "%s" has already been registered.' % args['name'])
return 1
if get_url_host(args['source']) is None:
logging.fatal('No repository handler for source URL "%s" found.' % args['source'])
return 1
dest = args['dest']
dest_host = None
dest_split = args['dest'].split(':')
if dest_split[0] in config['hosts']:
cfg_host = config['hosts'][dest_split[0]]
cfg_type = cfg_host['type'].lower()
hosts = load_hosts()
for cls in hosts.values():
if cls.TYPE != cfg_type:
continue
dest_host = cls(dest_split[0], cfg_host)
break
if dest_host is None:
logging.fatal('Host type "%s" not implemented.' % cfg_type)
return 1
try:
verifyObject(IHost, dest_host)
except BrokenImplementation:
logging.fatal('%r doesn\'t implement IHost correct.', dest_host)
return 1
repo = args['name']
if len(dest_split) > 1:
repo = dest_split[1]
url = dest_host.create_project(args['source'], repo)
if url == False:
return 1
dest = url
logging.info('Destination URL is: "%s".' % url)
if get_url_host(dest) is None:
logging.fatal('No repository handler for destination URL "%s" found.' % dest)
return 1
repo_cfg = {
'method': args['method'],
'source': args['source'],
'dest': dest,
'force': args['force'],
'prune': args['prune'],
}
config['repos'][args['name']] = repo_cfg
return 0

@ -0,0 +1,16 @@
import logging
from vcs_mirrors.lib.utils import sync_one
def configure_argparse(subparser):
subparser.add_argument('name',
help='Internal repository name')
def execute(config, args):
logging.info('Syncing "%s"' % args['name'])
if sync_one(args['name'], config) == False:
return 1
return 0

@ -0,0 +1,16 @@
import logging
from vcs_mirrors.lib.utils import sync_one
def configure_argparse(subparser):
pass
def execute(config, args):
result = 0
for repo in config['repos'].keys():
logging.info('Syncing "%s"' % repo)
if sync_one(repo, config) == False:
result = 1
return result

@ -0,0 +1,109 @@
import logging
from zope.interface import implementer
from vcs_mirrors.lib.interfaces import IHost
try:
from github import Github
from github.GithubException import GithubException
from github.GithubException import UnknownObjectException
GITHUB_AVAILABLE = True
except ImportError:
GITHUB_AVAILABLE = False
pass
__all__ = ['__virtual__', 'Host']
def __virtual__():
if not GITHUB_AVAILABLE:
logging.warn(
'Host type "github" isn\'t available, couldn\'t import pygithub')
return GITHUB_AVAILABLE
@implementer(IHost)
class Host(object):
TYPE = 'github'
_settings = None
def __init__(self, host, settings):
self._settings = {
'public': True,
'issues_enabled': True,
'wiki_enabled': True,
'downloads_enabled': True,
'projects_enabled': True,
'use_https': False,
}
self._settings.update(settings)
self._api = Github(self._settings['api_key'])
def create_project(self, source, repo):
desc = 'Git mirror of %s.' % source
if self._settings['public']:
desc = 'Public mirror of %s' % source
org, name = repo.split('/')
g_user = self._api.get_user()
logging.info('%s: Createing project: "%s"' % (self, repo))
g_repo = None
if org == g_user.login:
try:
g_repo = g_user.create_repo(
name,
description=desc,
private=not self._settings['public'],
has_issues=self._settings['issues_enabled'],
has_wiki=self._settings['wiki_enabled'],
has_downloads=self._settings['downloads_enabled'],
has_projects=self._settings['projects_enabled'],
)
except GithubException:
return False
else:
# Find org
try:
g_org = self._api.get_organization(org)
except UnknownObjectException:
return False
# Create repo
try:
g_repo = g_org.create_repo(
name,
description=desc,
private=not self._settings['public'],
has_issues=self._settings['issues_enabled'],
has_wiki=self._settings['wiki_enabled'],
has_downloads=self._settings['downloads_enabled'],
has_projects=self._settings['projects_enabled'],
)
except GithubException:
return False
if self._settings['use_https']:
return g_repo.clone_url
return g_repo.ssh_url
def get_url(self, repo):
g_repo = self._api.get_repo(repo)
if self._settings['use_https']:
return g_repo.clone_url
return g_repo.ssh_url
def __repr__(self):
return '<%s>' % self.TYPE
def __str__(self):
return '%s' % self.TYPE

@ -0,0 +1,141 @@
"""
Large parts from: https://github.com/samrocketman/gitlab-mirrors/blob/development/lib/manage_gitlab_project.py
"""
import logging
from zope.interface import implementer
from vcs_mirrors.lib.interfaces import IHost
try:
import gitlab
from gitlab.exceptions import GitlabCreateError
GITLAB_AVAILABLE = True
except ImportError:
GITLAB_AVAILABLE = False
pass
__all__ = ['__virtual__', 'Host']
def __virtual__():
if not GITLAB_AVAILABLE:
logging.warn(
'Host type "gitlab" isn\'t available, couldn\'t import python-gitlab')
return GITLAB_AVAILABLE
def _find_matches(objects, kwargs, find_all):
"""Helper function for _add_find_fn. Find objects whose properties
match all key, value pairs in kwargs.
Source: https://github.com/doctormo/python-gitlab3/blob/master/gitlab3/__init__.py
"""
ret = []
for obj in objects:
match = True
# Match all supplied parameters
for param, val in kwargs.items():
if not getattr(obj, param) == val:
match = False
break
if match:
if find_all:
ret.append(obj)
else:
return obj
if not find_all:
return None
return ret
@implementer(IHost)
class Host(object):
TYPE = 'gitlab'
_settings = None
def __init__(self, host, settings):
self._settings = {
'url': host,
'ssl_verify': True,
'public': False,
'issues_enabled': False,
'wall_enabled': False,
'merge_requests_enabled': False,
'wiki_enabled': False,
'snippets_enabled': False,
'use_https': False,
}
self._settings.update(settings)
# pylint: disable=E1101
self._api = gitlab.Gitlab(
'https://' + self._settings['url'],
self._settings['api_key'],
ssl_verify=self._settings['ssl_verify'],
api_version=4
)
# pylint: enable=E1101
def _find_group(self, **kwargs):
groups = self._api.groups.list()
return _find_matches(groups, kwargs, False)
def _find_project(self, **kwargs):
projects = self._api.projects.list(as_list=True)
return _find_matches(projects, kwargs, False)
def create_project(self, source, repo):
desc = 'Git mirror of %s.' % source
if self._settings['public']:
desc = 'Public mirror of %s' % source
group, name = repo.split('/')
group_obj = self._find_group(name=group)
if group_obj is None:
logging.info('%s: Createing group: "%s"' % (self, group))
group_obj = self._api.groups.create({'name': group, 'path': group})
project_options = {
'name': name,
'description': desc,
'issues_enabled': str(self._settings['issues_enabled']).lower(),
'wall_enabled': str(self._settings['wall_enabled']).lower(),
'merge_requests_enabled': str(self._settings['merge_requests_enabled']).lower(),
'wiki_enabled': str(self._settings['wiki_enabled']).lower(),
'snippets_enabled': str(self._settings['snippets_enabled']).lower(),
'namespace_id': group_obj.id
}
logging.info('%s: Createing project: "%s"' % (self, repo))
try:
project = self._api.projects.create(project_options)
except GitlabCreateError:
logging.error('Cannot create project "%s", error: path has already been taken.' % repo)
return False
if self._settings['use_https']:
return project.http_url_to_repo
return project.ssh_url_to_repo
def get_url(self, repo):
project = self._find_project(name=repo)
if project is None:
return None
if self._settings['use_https']:
return project.http_url_to_repo
return project.ssh_url_to_repo
def __repr__(self):
return '<%s(%s)>' % (self.TYPE, self._settings['url'])
def __str__(self):
return '%s:%s' % (self.TYPE, self._settings['url'])

@ -0,0 +1,16 @@
import os
class Cd:
"""Context manager for changing the current working directory
From: https://stackoverflow.com/a/13197763/3368468
"""
def __init__(self, newPath):
self.newPath = os.path.expanduser(newPath)
def __enter__(self):
self.savedPath = os.getcwd()
os.chdir(self.newPath)
def __exit__(self, etype, value, traceback):
os.chdir(self.savedPath)

@ -0,0 +1,11 @@
import ruamel.yaml
def load_config(path):
with open(path, 'r') as fp:
return ruamel.yaml.round_trip_load(fp.read())
def save_config(config, path):
with open(path, 'w') as fp:
ruamel.yaml.round_trip_dump(config, fp)

@ -0,0 +1,51 @@
from zope.interface import Attribute, Interface
# pylint: disable=E0239,E0213
class IHost(Interface):
"""
Host interface, all Hosts MUST implement this
"""
TYPE = Attribute("""Type of the Host (normaly lowercase of module name)""")
def create_project(source, repo):
"""
Create the repo "repo" on the host
Returns the URL for the created project or False on error.
:return: url or False
"""
def get_url(repo):
"""
Returns the URL for the given repo
and None if not found.
:return: url or None
"""
class IRepo(Interface):
"""
Repo interface, all Repos MUST implement this
"""
TYPE = Attribute("""Type of the Repo (normaly lowercase of module name)""")
def fetch(repo):
"""
Mirrors the source url to the internal store.
:param repo: Internal repo name
:return: boolean
"""
def push(repo):
"""
Push from the internal store to the destination.
:param repo: Internal repo name
:return: boolean
"""
# pylint: enable=E0239,E0213

@ -0,0 +1,71 @@
import importlib
import pkgutil
import logging
import vcs_mirrors.command
import vcs_mirrors.host
import vcs_mirrors.repo
from vcs_mirrors.lib.interfaces import IHost, IRepo
REPOS = None
HOSTS = None
COMMANDS = None
def _load_all(package):
"""
https://stackoverflow.com/a/1707786/3368468
"""
result = {}
for _, modname, ispkg in pkgutil.iter_modules(package.__path__):
if ispkg == False:
module = importlib.import_module(
package.__name__ + '.' + modname, package)
result[modname] = module
return result
def load_commands():
global COMMANDS
if COMMANDS is not None:
return COMMANDS
COMMANDS = _load_all(vcs_mirrors.command)
return COMMANDS
def load_repos():
global REPOS
if REPOS is not None:
return REPOS
REPOS = {}
for k, m in _load_all(vcs_mirrors.repo).items():
# pylint: disable=E1120
if IRepo.implementedBy(m.Repo):
REPOS[k] = m.Repo
else:
logging.error('Repo "%s" doesn\'t implement IRepo' % k)
# pylint: enable=E1120
return REPOS
def load_hosts():
global HOSTS
if HOSTS is not None:
return HOSTS
HOSTS = {}
for k, m in _load_all(vcs_mirrors.host).items():
if not m.__virtual__():
continue
# pylint: disable=E1120
if IHost.implementedBy(m.Host):
HOSTS[k] = m.Host
else:
logging.error('Host "%s" doesn\'t implement IHost' % k)
# pylint: enable=E1120
return HOSTS

@ -0,0 +1,64 @@
import logging
import subprocess
from vcs_mirrors.lib.loader import load_repos
def get_url_host(url):
repos = load_repos()
for repo in repos.values():
host = repo.get_host(url)
if host is not None:
return host
return None
def get_repo_for_url(url, config):
repos = load_repos()
result = None
for repo in repos.values():
host = repo.get_host(url)
if host is not None:
result = repo(config)
break
return result
def sync_one(name, config):
if name not in config['repos']:
logging.fatal('Unknown repository "%s" given.' % name)
return False
repo_config = config['repos'][name]
source_repo = get_repo_for_url(repo_config['source'], config)
if source_repo is None:
logging.fatal(
'No repository handler for source URL "%s" found.' % repo_config['source'])
return False
dest_repo = get_repo_for_url(repo_config['dest'], config)
if dest_repo is None:
logging.fatal(
'No repository handler for source URL "%s" found.' % repo_config['dest'])
return False
if source_repo.fetch(name) == False:
return False
if dest_repo.push(name) == False:
return False
return True
def run_cmd(args, log_error=True):
logging.debug('Run: %s' % " ".join(args))
result = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if log_error:
if result.returncode != 0:
logging.error(result.stderr.decode('unicode_escape'))
return result

@ -0,0 +1,35 @@
import re
from vcs_mirrors.lib.utils import run_cmd
from vcs_mirrors.repo.git import Repo as GitRepo
class Repo(GitRepo):
TYPE = 'bazaar'
@staticmethod
def get_host(url):
"""
Returns the hostname when the url is a git host else None
Tested urls:
- bzr::bzr://bzr.savannah.gnu.org/emacs/trunk
- bzr::bzr://bzr.savannah.gnu.org/emacs
- bzr::lp:ubuntu/hello
- bzr::lp:bzr
- bzr::sftp://bill@mary-laptop/cool-repo/cool-trunk
"""
url = str(url).lower()
match = re.search(r'bzr::([-\d\w_\.]+):(//)?([\-0-9a-z_\.]+@+)?([\-0-9a-z_\.]+)(/.+)?', url)
if match:
if match.group(1) == 'lp':
return 'lp'
else:
return match.group(2)
return None
def _after_clone(self):
run_cmd(['git', 'gc', '--aggressive'])
super()._after_clone()

@ -0,0 +1,138 @@
import logging
import os
import os.path
import re
import shlex
import subprocess
from zope.interface import implementer
from vcs_mirrors.lib.cd import Cd
from vcs_mirrors.lib.interfaces import IRepo
from vcs_mirrors.lib.utils import run_cmd
__all__ = ['Repo']
@implementer(IRepo)
class Repo(object):
TYPE = 'git'
@staticmethod
def get_host(url):
"""
Returns the hostname when the url is a git host else None
Tested urls:
- https://github.com/pcdummy/vcs-mirrors.git
- https://git.lxch.eu/vcs-mirrors.git
- ssh://user@server/project.git
- user@server:project.git
- git://git.proxmox.com/git/aab.git
"""
url = str(url)
# https://github.com/pcdummy/vcs-mirrors.git
# https://git.lxch.eu/vcs-mirrors.git
match = re.search(r'https://([-\d\w_\.]+)/.*\.git$', url)
if match:
return match.group(1)
# ssh://user@server/project.git
match = re.search(
r'(ssh://)?[-\d\w_\.]+@{1}([-\d\w_\.]+)[:/]{1}.*\.git$', url)
if match:
return match.group(2)
# git://git.lxch.eu/git/aab.git
match = re.search(r'git://([-\d\w_\.]+)/.*\.git$', url)
if match:
return match.group(1)
return None
_config = None
def _after_clone(self):
pass
def __init__(self, config):
self._config = config
def fetch(self, repo):
logging.debug('%s: Fetching repository "%s"' % (self, repo))
repo_config = self._config['repos'][repo]
force = False
if 'force' in repo_config:
force = repo_config['force']
prune = False
if 'prune' in repo_config:
prune = repo_config['prune']
repo_dir = os.path.join(self._config['settings']['local_path'], repo)
repo_dir_exists = True
if not os.path.exists(repo_dir):
repo_dir_exists = False
logging.debug('Makedirs: "%s"', repo_dir)
os.makedirs(repo_dir, mode=0o700)
with Cd(repo_dir):
if not repo_dir_exists:
run_cmd(['git', 'clone', '--bare', shlex.quote(repo_config['source']), '.'])
run_cmd(['git', 'remote', 'rename', 'origin', 'source'])
self._after_clone()
else:
has_repo = run_cmd(['git', 'remote', 'get-url', 'source'], False).returncode == 0
if not has_repo:
run_cmd(['git', 'remote', 'add', 'source', shlex.quote(repo_config['source'])])
args = ['git', 'fetch']
if force:
args.append('--force')
if prune:
args.append('--prune')
args.append('source')
run_cmd(args)
return True
def push(self, repo):
logging.debug('%s: Pushing to repository "%s"' % (self, repo))
repo_dir = os.path.join(self._config['settings']['local_path'], repo)
if not os.path.exists(repo_dir):
logging.error('%s: Can\'t push repo "%s", it does not exists localy.' % (self, repo))
repo_config = self._config['repos'][repo]
force = False
if 'force' in repo_config:
force = repo_config['force']
prune = False
if 'prune' in repo_config:
prune = repo_config['prune']
with Cd(repo_dir):
has_repo = run_cmd(['git', 'remote', 'get-url', 'dest'], False).returncode == 0
if not has_repo:
run_cmd(['git', 'remote', 'add', 'dest', shlex.quote(repo_config['dest'])])
args = ['git', 'push']
if force:
args.append('--force')
if prune:
args.append('--prune')
args.extend(['dest', '--all'])
run_cmd(args)
return True
def __str__(self):
return self.TYPE
def __repr__(self):
return '<repo(%s)>' % self.TYPE

@ -0,0 +1,68 @@
import argparse
import os
import sys
import logging
from vcs_mirrors.lib.config import load_config
from vcs_mirrors.lib.config import save_config
from vcs_mirrors.lib.loader import load_commands
DEFAULT_CONFIG_FILE = os.getenv('VCS_MIRROR_CONFIG', 'vcs-mirrors.yaml')
DEFAULT_LOG_LEVEL = logging.INFO
DEFAULT_LOG_FORMAT = '%(levelname)-8.8s %(message)s'
def main(args=None):
"""The main routine."""
if args is None:
args = sys.argv[1:]
parser = argparse.ArgumentParser(
description='vcs-mirror Command-Line Interface')
subparsers = parser.add_subparsers(title='subcommands',
description='Main vcs-mirror CLI commands',
dest='subcommand',
help='Choose and run with --help')
subparsers.required = True
cmds = load_commands()
for name, cmd in cmds.items():
subparser = subparsers.add_parser(name)
subparser.set_defaults(which=name)
subparser.add_argument('--config',
help='Application configuration file',
dest='yaml_file',
required=False,
default=DEFAULT_CONFIG_FILE)
subparser.add_argument('-q', '--quiet', action='store_const',
const=logging.CRITICAL, dest='verbosity',
help='Show only critical errors.')
subparser.add_argument('-v', '--debug', action='store_const',
const=logging.DEBUG, dest='verbosity',
help='Show all messages, including debug messages.')
cmd.configure_argparse(subparser)
# Parse command-line arguments
parsed_args = vars(parser.parse_args(args))
# Initialize logging from
level = parsed_args.get('verbosity') or DEFAULT_LOG_LEVEL
logging.basicConfig(level=level, format=DEFAULT_LOG_FORMAT)
config = load_config(parsed_args['yaml_file'])
# Execute the command
which_command = parsed_args['which']
result = cmds[which_command].execute(config, parsed_args)
if result != 0:
sys.exit(result)
save_config(config, parsed_args['yaml_file'])
Loading…
Cancel
Save