in src/lambda_codebase/initial_commit/bootstrap_repository/adf-build/shared/cdk/cdk_constructs/adf_codepipeline.py [0:0]
def _generate_configuration(self): #pylint: disable=R0912, R0911, R0915
if self.provider == "Manual" and self.category == "Approval":
_props = {
"CustomData": self.target.get('properties', {}).get('message') or f"Approval stage for {self.map_params['name']}"
}
if self.notification_endpoint:
_props["NotificationArn"] = self.notification_endpoint
if self.target.get('properties', {}).get('sns_topic_arn'):
_props["NotificationArn"] = self.target.get('properties', {}).get('sns_topic_arn')
return _props
if self.provider == "S3" and self.category == "Source":
return {
"S3Bucket": self.map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('bucket_name'),
"S3ObjectKey": self.map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('object_key'),
"PollForSourceChanges": self.map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('trigger_on_changes', True),
}
if self.provider == "S3" and self.category == "Deploy":
return {
"BucketName": self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get(
'bucket_name') or self.target.get(
'properties', {}).get(
'bucket_name'),
"Extract": self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get(
'extract') or self.target.get(
'properties', {}).get(
'extract', False),
"ObjectKey": self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get(
'object_key') or self.target.get(
'properties', {}).get(
'object_key')
}
if self.provider == "CodeStarSourceConnection":
owner = self.map_params.get('default_providers', {}).get('source').get('properties', {}).get('owner', {})
repo = self.map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('repository', {}) or self.map_params['name']
codestar_connection_path = self.map_params.get('default_providers', {}).get('source').get('properties', {}).get('codestar_connection_path', {})
ssm_client = boto3.client('ssm')
try:
response = ssm_client.get_parameter(Name=codestar_connection_path)
except Exception as e:
LOGGER.error(f"No parameter found at {codestar_connection_path}. Check the path/value.")
raise e
connection_arn = response['Parameter']['Value']
return {
"ConnectionArn": connection_arn,
"FullRepositoryId": f"{owner}/{repo}",
"BranchName": self.map_params.get('default_providers', {}).get(
'source', {}).get('properties', {}).get(
'branch',
self.default_scm_branch
)
}
if self.provider == "GitHub":
return {
"Owner": self.map_params.get('default_providers', {}).get('source').get('properties', {}).get('owner', {}),
"Repo": self.map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('repository', {}) or self.map_params['name'],
"Branch": self.map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('branch', self.default_scm_branch),
# pylint: disable=no-value-for-parameter
"OAuthToken": core.SecretValue.secrets_manager(
self.map_params['default_providers']['source'].get('properties', {}).get('oauth_token_path'),
json_field=self.map_params['default_providers']['source'].get('properties', {}).get('json_field')
),
"PollForSourceChanges": False
}
if self.provider == "Lambda":
return {
"FunctionName": self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get(
'function_name', '') or self.target.get(
'properties', {}).get(
'function_name', ''),
"UserParameters": str(self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get('input', '') or self.target.get(
'properties', {}).get(
'input', ''))
}
if self.provider == "CloudFormation":
_path_prefix = self.target.get(
'properties', {}).get(
'root_dir') or self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get(
'root_dir') or ""
if _path_prefix and not _path_prefix.endswith('/'):
_path_prefix = f"{_path_prefix}/"
_input_artifact = f"{self.map_params['name']}-build"
_props = {
"ActionMode": self.action_mode,
"StackName": (
self.target.get('properties', {}).get('stack_name')
or self.map_params.get('default_providers', {}).get(
'deploy', {}).get('properties', {}).get('stack_name')
or f"{ADF_STACK_PREFIX}{self.map_params['name']}"
),
"ChangeSetName": f"{ADF_STACK_PREFIX}{self.map_params['name']}",
"TemplateConfiguration": (
f"{_input_artifact}::{_path_prefix}params/{self.target['name']}_{self.region}.json"
),
"Capabilities": "CAPABILITY_NAMED_IAM,CAPABILITY_AUTO_EXPAND",
"RoleArn": self.role_arn if self.role_arn else (
f"arn:{ADF_DEPLOYMENT_PARTITION}:iam::{self.target['id']}:"
f"role/adf-cloudformation-deployment-role"
)
}
if self.map_params.get('default_providers', {}).get('build', {}).get('properties', {}).get('environment_variables', {}).get('CONTAINS_TRANSFORM'):
_props["TemplatePath"] = f"{_input_artifact}::{_path_prefix}template_{self.region}.yml"
else:
_template_filename = self.target.get(
'properties', {}).get(
'template_filename') or self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get(
'template_filename') or "template.yml"
_props["TemplatePath"] = f"{_input_artifact}::{_path_prefix}{_template_filename}"
if self.target.get('properties', {}).get('outputs'):
_props['OutputFileName'] = f"{_path_prefix}{self.target['properties']['outputs']}.json"
if self.target.get('properties', {}).get('param_overrides'):
_overrides = {}
for override in self.target.get('properties', {}).get('param_overrides', []):
_overrides[override['param']] = {
"Fn::GetParam": [
override['inputs'],
f"{override['inputs']}.json",
override['key_name'],
],
}
_props['ParameterOverrides'] = json.dumps(_overrides)
return _props
if self.provider == "Jenkins":
return {
"ProjectName": self.map_params['default_providers']['build'].get(
'properties', {}).get(
'project_name', self.map_params['name']), # Enter the name of the project you created in the Jenkins plugin
"ServerURL": self.map_params['default_providers']['build'].get('properties', {}).get('server_url'), # Server URL
"ProviderName": self.map_params['default_providers']['build'].get('properties', {}).get('provider_name') # Enter the provider name you configured in the Jenkins plugin
}
if self.provider == "CodeBuild":
return {
"ProjectName": self.project_name or f"adf-build-{self.map_params['name']}"
}
if self.provider == "ServiceCatalog":
return {
"ConfigurationFilePath": (
self.target.get('properties', {}).get('configuration_file_path')
or f"params/{self.target['name']}_{self.region}.json"
),
"ProductId": self.target.get(
'properties', {}).get(
'product_id') or self.map_params['default_providers']['deploy'].get(
'properties', {}).get(
'product_id') # product_id is required for Service Catalog, meaning the product must already exist.
}
if self.provider == "CodeDeploy":
return {
"ApplicationName": self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get(
'application_name', {}) or self.target.get(
'properties', {}).get(
'application_name'),
"DeploymentGroupName": self.map_params.get(
'default_providers', {}).get(
'deploy', {}).get(
'properties', {}).get(
'deployment_group_name', {}) or self.target.get(
'properties', {}).get(
'deployment_group_name')
}
if self.provider == "CodeCommit":
props = {
"BranchName": self.map_params['default_providers']['source'].get('properties', {}).get('branch', self.default_scm_branch),
"RepositoryName": self.map_params['default_providers']['source'].get('properties', {}).get('repository', {}) or self.map_params['name'],
"PollForSourceChanges": (
self.map_params['default_providers']['source'].get('properties', {}).get('trigger_on_changes', True)
and self.map_params['default_providers']['source'].get('properties', {}).get('poll_for_changes', False)
)
}
output_artifact_format = self.map_params['default_providers']['source'].get('properties', {}).get('output_artifact_format', None)
if output_artifact_format:
props["OutputArtifactFormat"] = output_artifact_format
return props
raise Exception(f"{self.provider} is not a valid provider")