Source code for aeriscloud.project
import os
from slugify import slugify
from .box import Box, BoxList
from .config import projects_path, aeriscloud_path
from .log import get_logger
from .utils import jinja_env
from .vagrant import run
logger = get_logger('project')
BASEBOX_DEFAULT_CPU = 1
BASEBOX_DEFAULT_RAM = 1024
# caches the project list but allows reloading the list
# TODO: extend dict and provide the right helpers
class ProjectList(object):
def __init__(self):
self.project_list = {}
def load(self):
self.project_list = {}
search_path = projects_path()
if not os.path.isdir(search_path):
return
sub_files = [os.path.join(search_path, name)
for name in os.listdir(search_path)]
for project_dir in sub_files:
if not os.path.isdir(project_dir):
continue
if not os.path.isfile(os.path.join(project_dir,
'.aeriscloud.yml')):
continue
project = Project(project_dir)
if project._initialized and project.name() \
and project.name().lower() not in self.project_list:
self.project_list[project.name().lower()] = project
def __call__(self):
# lazy loading
if not self.project_list:
self.load()
return self.project_list
[docs]def all():
"""
Return every projects available on the local host
:return: list[Project]
"""
return projects().values()
[docs]def get(name):
"""
Return the project corresponding to the given project name or directory
:param name: str
:return: Project|None
"""
project_list = projects()
if name in project_list:
return project_list[name]
if os.path.isdir(os.path.join(projects_path(), name)):
return Project(os.path.join(projects_path(), name))
return None
def from_cwd():
path = os.getcwd()
pro_path = projects_path()
# walk path upward until we find a .aeriscloud.yml file or until we
# reach the project folder
while True:
pro = from_path(path)
if pro:
return pro
if path == pro_path or not path.startswith(pro_path):
break
path = os.path.dirname(path)
return None
[docs]def from_path(path):
"""
Return the project stored in the given folder if any
:return: Project|None
"""
if not os.path.isfile(os.path.join(path, '.aeriscloud.yml')):
return None
return Project(path)
[docs]class Project(object):
"""
Represents an AerisCloud project
:param folder: str The folder where the project is stored
"""
def __init__(self, folder):
self._folder = folder
self._config_file = os.path.join(self._folder, '.aeriscloud.yml')
self._initialized = False
self._config = None
self._boxes = BoxList()
self._logger = get_logger(self.folder(True), logger)
if os.path.isfile(self._config_file):
self._initialized = True
def _load_config(self):
if not os.path.isfile(self._config_file):
self._config = {}
return
import yaml
with open(self._config_file) as fd:
self._config = yaml.load(fd) or {}
def _load_infra(self):
if self._boxes:
return
# We store boxes in a list as python dicts are not ordered
if 'boxes' in self.config():
self._boxes = BoxList([Box(self, box)
for box in self.config()['boxes']])
def initialized(self):
return self._initialized
[docs] def name(self):
"""
Return the name of the project
:return: str
"""
if not self._config or 'project_name' not in self.config():
return self.folder(True)
return self.config()['project_name']
def set_name(self, name):
self.config()['project_name'] = name
self._logger = get_logger(self.name(), logger)
def get_production_url(self):
if 'production_url' not in self.config():
return None
return self.config()['production_url']
def id(self):
if 'id' not in self.config():
return None
return self.config()['id']
def set_id(self, project_id):
self.config()['id'] = project_id
def organization(self):
if 'organization' not in self.config():
return None
return self.config()['organization']
def set_organization(self, organization):
self.config()['organization'] = organization
def services(self):
if 'services' not in self.config():
return []
return self.config()['services']
def set_services(self, services):
self.config()['services'] = services
def add_box(self, basebox, basebox_url=None):
if 'boxes' not in self.config():
self.config()['boxes'] = []
if 'basebox' not in basebox:
raise ValueError('Invalid basebox dict provided')
basebox.setdefault('name', slugify(basebox['basebox']))
basebox.setdefault('cpu', BASEBOX_DEFAULT_CPU)
basebox.setdefault('ram', BASEBOX_DEFAULT_RAM)
if basebox in self.config()['boxes']:
return
if basebox_url:
self.config()['basebox_url'] = basebox_url
self.config()['boxes'].append(basebox)
self._load_infra()
def folder(self, base=False):
if base:
return os.path.basename(self._folder)
return self._folder
def config(self):
if self._config is None:
self._load_config()
self._load_infra()
return self._config
def config_file(self):
return self._config_file
[docs] def boxes(self):
"""
Return the list of boxes available for this project
:return: BoxList[Box]
"""
self._load_infra()
return self._boxes
def rsync_enabled(self):
return 'use_rsync' in self.config() and self.config()['use_rsync']
[docs] def box(self, name=''):
"""
Retrieve a box by name
:param name: str
:return: Box|None
"""
self._load_infra()
if not self._boxes:
return None
if not name:
return self.boxes()[0]
for box in self._boxes:
if box.name() == name:
return box
return None
def save(self):
self._logger.info('writing .aeriscloud.yml')
env = jinja_env()
env.filters['yaml'] = _yaml_filter
config = env.get_template('aeriscloud.yml.j2') \
.render(config=_ProjectConfig(self.config()))
with open(self._config_file, 'w') as fd:
fd.write(config)
self._initialized = True
def vagrant(self, *args, **kwargs):
self._logger.info('running: vagrant %s', ' '.join(args))
return run(self, *args, **kwargs)
def vagrant_dir(self):
return os.path.join(self.folder(), '.vagrant')
def vagrant_working_dir(self):
return aeriscloud_path
def endpoints(self):
if 'browse' in self.config():
return self.config()['browse']
else:
return {}
def __repr__(self):
return '<Project %s [%s]>' % (self.name(), self._folder)
def _yaml_filter(val, name=None):
import yaml
if name:
return yaml.dump({name: val}, default_flow_style=False)
return yaml.dump(val, default_flow_style=False)
class _ProjectConfig():
def __init__(self, config):
self._config = config
self._accessed = []
def __getattr__(self, key):
if key not in self._config:
return None
if key not in self._accessed:
self._accessed.append(key)
return self._config[key]
def extra(self):
return dict([(key, item) for key, item in self._config.iteritems()
if key not in self._accessed])
# load projects on import
projects = ProjectList()