Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
generate terraform apigateway resources from oas3.0 json export
#!/usr/local/bin/python
# this script will export the API gateway config as an Open API 3.0 json blob
# and then create the appropriate terraform resources
import json
from subprocess import run as sp_run
import boto3
profile_name = 'dev'
output_file = 'terraform/regional/api_gw_methods.tf'
client = boto3.Session(profile_name=profile_name, region_name='us-east-1').client('apigateway')
def get_api_id(api_name):
response = client.get_rest_apis()
for apigw in response['items']:
if apigw['name'] == api_name:
return apigw['id']
return False
response = client.get_export(
restApiId=get_api_id('userlink'),
stageName='v1',
exportType='oas30',
parameters={
'extensions': 'integrations'
},
accepts = "application/json"
)
json_data = json.loads(response['body'].read().decode('utf-8').replace('/v1/', '/'))
# this logic finds paths not explicitly defined in the api definition but we still need parent terraform resources created for
all_paths = []
explicit_paths = [path.replace("/", "", 1) for path in json_data["paths"]]
for path in json_data["paths"]:
path_name = path.replace("/", "_").replace("_","",1)
split_path = path_name.split("_")
for i in range(len(split_path)):
current = "/".join(split_path[:i+1])
if current not in all_paths: all_paths.append(current)
for path in all_paths:
if path not in explicit_paths:
json_data["paths"][f'/{path}'] = {}
# we do this to get the terraform resources grouped together properly
json_data = json.dumps(json_data, sort_keys=True)
json_data = json.loads(json_data)
output = []
for path in json_data["paths"]:
path_name = path.replace("{", "").replace("}", "").replace("/", "_")
output.append(f'resource "aws_api_gateway_resource" "{path_name}" {{')
output.append("rest_api_id = aws_api_gateway_rest_api.userlink.id")
output.append(
f'parent_id = aws_api_gateway_resource.{"_".join(path_name.split("_")[:-1])}.id'
if len(path_name.split("_")[:-1]) > 1
else "parent_id = aws_api_gateway_resource._stage_version.id"
)
output.append(f'path_part = "{path.split("/")[-1]}"')
output.append("}\n")
for method in json_data["paths"][path]:
method_type = (
"ANY" if method == "x-amazon-apigateway-any-method" else method.upper()
)
output.append(f'resource "aws_api_gateway_method" "{path_name}__{method_type}" {{')
output.append("rest_api_id = aws_api_gateway_rest_api.userlink.id")
output.append(f"resource_id = aws_api_gateway_resource.{path_name}.id")
output.append(f'http_method = "{method_type}"')
output.append(
"authorization = "
+ (
'"NONE" #tfsec:ignore:aws-api-gateway-no-public-access'
if not json_data["paths"][path][method].get("security")
else '"CUSTOM"\n authorizer_id = aws_api_gateway_authorizer.auth.id'
)
)
if json_data["paths"][path][method].get("parameters"):
output.append("\n request_parameters = {")
params = []
for parameters in json_data["paths"][path][method]["parameters"]:
params.append(
f'"method.request.{parameters["in"]}.{parameters["name"]}" = {str(parameters["required"]).lower()}'
)
for param in sorted(params):
output.append(param)
output.append(" }")
output.append("}\n")
if json_data["paths"][path][method].get("x-amazon-apigateway-integration"):
output.append(
f'resource "aws_api_gateway_integration" "{path_name}__{method_type}_userlink" {{'
)
output.append(f"rest_api_id = aws_api_gateway_rest_api.userlink.id")
output.append(f"resource_id = aws_api_gateway_resource.{path_name}.id")
output.append(
f" http_method = aws_api_gateway_method.{path_name}__{method_type}.http_method"
)
output.append(f'integration_http_method = "POST"')
output.append(f'type = "AWS_PROXY"')
output.append(f"uri = local.lambda_invoke_arn")
output.append("}\n")
for response in json_data["paths"][path][method]["responses"]:
output.append(
f'resource "aws_api_gateway_method_response" "{path_name}__{method_type}__{response}" {{'
)
output.append("rest_api_id = aws_api_gateway_rest_api.userlink.id")
output.append(f"resource_id = aws_api_gateway_resource.{path_name}.id")
output.append(
f"http_method = aws_api_gateway_method.{path_name}__{method_type}.http_method"
)
output.append(f"status_code = {response}")
output.append("}\n")
with open(output_file, "w") as outfile:
for line in output:
outfile.write(f'{line}\n')
sp_run(["terraform", "fmt", output_file])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment