Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def debug_flag():
"""Set logging level for entry points."""
logging.basicConfig(format=consts.LOGGING_FORMAT)
parser = argparse.ArgumentParser(description=debug_flag.__doc__)
add_debug(parser)
args, _extra_args = parser.parse_known_args()
package, *_ = __package__.split('.')
logging.getLogger(package).setLevel(args.debug)
def create_pipeline(self, onetime=None):
"""Create the spinnaker pipeline(s)."""
utils.banner("Creating Pipeline")
kwargs = {
'app': self.app,
'trigger_job': self.trigger_job,
'prop_path': self.json_path,
'base': None,
'runway_dir': self.runway_dir,
}
pipeline_type = self.configs['pipeline']['type']
if pipeline_type not in consts.ALLOWED_TYPES and pipeline_type not in consts.MANUAL_TYPES:
raise NotImplementedError('Pipeline type "{0}" not permitted.'.format(pipeline_type))
if not onetime:
if pipeline_type == 'lambda':
spinnakerpipeline = pipeline.SpinnakerPipelineLambda(**kwargs)
elif pipeline_type == 's3':
spinnakerpipeline = pipeline.SpinnakerPipelineS3(**kwargs)
elif pipeline_type == 'datapipeline':
spinnakerpipeline = pipeline.SpinnakerPipelineDataPipeline(**kwargs)
elif pipeline_type in consts.MANUAL_TYPES:
spinnakerpipeline = pipeline.SpinnakerPipelineManual(**kwargs)
else:
# Handles all other pipelines
spinnakerpipeline = pipeline.SpinnakerPipeline(**kwargs)
else:
spinnakerpipeline = pipeline.SpinnakerPipelineOnetime(onetime=onetime, **kwargs)
debug_flag()
self.email = os.getenv("EMAIL")
self.env = os.getenv("ENV")
self.group = os.getenv("PROJECT")
self.region = os.getenv("REGION")
self.repo = os.getenv("GIT_REPO")
self.runway_dir = os.getenv("RUNWAY_DIR")
self.artifact_path = os.getenv("ARTIFACT_PATH")
self.artifact_version = os.getenv("ARTIFACT_VERSION")
self.promote_stage = os.getenv("PROMOTE_STAGE", "latest")
self.provider = os.getenv("PROVIDER", "aws")
self.git_project = "{}/{}".format(self.group, self.repo)
parsed = gogoutils.Parser(self.git_project)
generated = gogoutils.Generator(*parsed.parse_url(), formats=consts.APP_FORMATS)
self.app = generated.app_name()
self.trigger_job = generated.jenkins()['name']
self.git_short = generated.gitlab()['main']
self.raw_path = "./raw.properties"
self.json_path = self.raw_path + ".json"
self.configs = None
#!/usr/bin/env python
""" A runner for all of the spinnaker pipe modules
Looks for environment variables from Jenkins and then
runs specific prepare jobs
"""
import logging
import os
import gogoutils
from foremast import (configs, consts, iam, s3, securitygroup, elb, dns,
slacknotify, app, pipeline, utils)
LOG = logging.getLogger(__name__)
logging.basicConfig(format=consts.LOGGING_FORMAT)
logging.getLogger("foremast").setLevel(logging.INFO)
class ForemastRunner:
""" Wraps each pipes module in a way that is easy to invoke """
def __init__(self):
""" Setups the Runner for all Foremast modules
Args:
group (str): Gitlab group of the application
repo (str): The application repository name
env (str): Deployment environment, dev/stage/prod etc
region (str): AWS region (us-east-1)
"""
# get environment variables. Defaults to None of none found
self.group = os.getenv("PROJECT")