import json
import os
import re
import pkg_resources
from vee.config import Config
from vee.database import Database
from vee.devpackage import DevPackage
from vee.environmentrepo import EnvironmentRepo
from vee.git import GitRepo
from vee.utils import makedirs, cached_property, find_home, DB_NAME
from vee import log
# We shall call the default repository "primary", as it is a nice generic name
# and it does not start with any other letters in the path:
# $VEE/environments/primary/refs/origin/master
PRIMARY_REPO = 'primary'
[docs]class Home(object):
"""The starting point of everything VEE.
:param str root: The root directory of the home; defaults to ``$VEE``.
:param str repo: The name of the default repository; defaults to ``$VEE_REPO``.
"""
def __init__(self, root=None, repo=None):
if root is None:
root = find_home()
if root is None:
raise ValueError('Need a root, or $VEE.')
self.root = root
self.default_repo_name = repo if repo is not None else os.environ.get('VEE_REPO')
dbpath = self._abs_path(DB_NAME)
self.db = Database(dbpath)
self.config = Config(self)
@property
def exists(self):
return self.db.exists
def init(self, if_not_exists=False, create_parents=False):
self._makedirs(create_parents) # Do this anyways.
if self.exists:
if if_not_exists:
return
raise ValueError('home already exists')
self.db.create()
@cached_property
def dev_root(self):
env_value = os.environ.get('VEE_DEV')
return os.path.expanduser(env_value) if env_value is not None else self._abs_path('dev')
@cached_property
def dev_search_path(self):
env_value = os.environ.get("VEE_DEV_PATH")
path = env_value.split(':') if env_value is not None else [self.dev_root]
path = [os.path.expanduser(x) for x in path]
return path
def _abs_path(self, *args):
return os.path.abspath(os.path.join(self.root, *args))
def _makedirs(self, create_parents=False):
if not create_parents and not os.path.exists(os.path.dirname(self.root)):
raise ValueError('parent of %s does not exist' % self.root)
for name in ('builds', 'environments', 'installs', 'packages', 'repos'):
path = self._abs_path(name)
makedirs(path)
def iter_development_packages(self, exists=True, search=True):
if not search:
# We used to have a development_packages table, and searching was
# something that had to be requested.
log.debug("iter_development_packages(..., seach=False) is deprecated.")
for root in self.dev_search_path:
if not os.path.exists(root):
continue
for name in os.listdir(root):
path = os.path.join(root, name)
if name.endswith('.vee-dev.json'):
yield DevPackage.from_tag(path, home=self)
continue
# Not used yet.
sub_path = os.path.join(path, '.vee-dev.json')
if os.path.exists(sub_path):
yield DevPackage.from_tag(sub_path, home=self)
def find_development_package(self, name):
if '/' in name:
paths = [name]
name = os.path.basename(name)
else:
paths = []
for root in self.dev_search_path:
path = os.path.join(root, name)
if os.path.exists(path):
paths.append(path)
for path in paths:
sidecar_path = os.path.join(os.path.dirname(path), '.' + name + '.vee-dev.json')
if os.path.exists(sidecar_path):
return DevPackage.from_tag(sidecar_path, home=self)
# Not used yet.
subcar_path = os.path.join(path, '.vee-dev')
if os.path.exists(subcar_path):
return DevPackage.from_tag(sidecar_path, home=self)
def iter_env_repos(self):
for row in self.db.execute('SELECT * FROM repositories'):
env_repo = EnvironmentRepo(row, home=self)
if env_repo.exists:
yield env_repo
def get_env_repo(self, name=None):
name = name or self.default_repo_name
if name:
row = self.db.execute('SELECT * FROM repositories WHERE name = ? LIMIT 1', [name]).fetchone()
if not row:
raise ValueError('%r repo does not exist' % name)
else:
# Grab the default repo if possible, otherwise make sure there is
# only one.
rows = self.db.execute('SELECT * FROM repositories ORDER BY is_default DESC LIMIT 2').fetchall()
if not rows:
raise ValueError('no repositories exist')
elif rows[0]['is_default']:
row = rows[0]
elif len(rows) == 1:
row = rows[0]
else:
raise ValueError('multiple repositories with no default')
env_repo = EnvironmentRepo(row, home=self)
if not env_repo.exists:
log.debug('Looking for env_repo: %s' % env_repo.work_tree)
raise ValueError('%r repo does not exist' % env_repo.name)
return env_repo
def create_env_repo(self, path=None, url=None, name=None, remote=None, branch=None, is_default=None):
if path:
path = os.path.abspath(path)
if not os.path.exists(path):
raise ValueError('no repo at %s' % path)
if url or path:
name = name or re.sub(r'\.git$', '', os.path.basename(url or path))
else:
name = name or self.default_repo_name or PRIMARY_REPO
# Make sure it doesn't exist.
try:
env_repo = self.get_env_repo(name)
except ValueError:
pass
else:
raise ValueError('%r repo already exists' % name)
con = self.db.connect()
cur = con.execute('INSERT OR REPLACE INTO repositories (name, path, remote, branch, is_default) VALUES (?, ?, ?, ?, ?)', [
name, path, remote or 'origin', branch or 'master', bool(is_default)])
row = con.execute('SELECT * FROM repositories WHERE id = ?', [cur.lastrowid]).fetchone()
env_repo = EnvironmentRepo(row, home=self)
if url:
env_repo.clone_if_not_exists(url)
elif not env_repo.exists:
makedirs(env_repo.work_tree)
env_repo.git('init')
return env_repo
def update_env_repo(self, name, url=None, remote=None, branch=None, is_default=None):
if not (url or remote or branch or is_default):
raise ValueError('provide something to update')
env_repo = self.get_env_repo(name)
if remote or branch or is_default:
env_repo.remote_name = remote or env_repo.remote_name
env_repo.branch_name = branch or env_repo.branch_name
self.db.execute('UPDATE repositories SET remote = ?, branch = ?, is_default = ? WHERE id = ?', [
env_repo.remote_name,
env_repo.branch_name,
int(bool(is_default or row['is_default'])),
env_repo.id,
])
if url:
env_repo.remotes(**{env_repo.remote_name: url})
def main(self, args, environ=None, **kwargs):
from vee.commands.main import main
environ = (environ or os.environ).copy()
environ['VEE'] = self.root
return main(args, environ, **kwargs)