bedrock.generators.generate-websockets-for-terraform

This script generates the AWS Websocket Configuration for the Bedrock application

 1"""
 2This script generates the AWS Websocket Configuration for the Bedrock application
 3"""
 4
 5import json
 6import sys
 7from pathlib import Path
 8
 9from bedrock._helpers.string import camelCase_to_snake_case, camelCase_to_kebab_case
10from bedrock.config._modules import _list_files_with_extensions
11
12BASE_PATH = None
13EXTENSION_RUNTIME_MAP = {
14    "py": {"language": "python", "runtime": "python3.13"},
15    # Add more as needed
16}
17
18
19def make_websockets_json(websocket_instances):
20    """
21    Runs through each websocket and generates a JSON object with the websocket name.
22    Result example:
23    ```json
24    {
25          "connect": {
26            "name": "connect",
27            "type": "python",
28            "runtime": "python3.13",
29            "zipName": "connect.zip"
30          }
31    }
32    ```
33    """
34    websockets = {}
35    for websocket in websocket_instances:
36        filename = websocket[0]
37        extension = websocket[1]
38        snake_case_name = camelCase_to_snake_case(filename)
39        kebab_case_name = camelCase_to_kebab_case(filename)
40
41        route_key = kebab_case_name
42        if route_key in {"connect", "disconnect", "default"}:
43            route_key = f"${kebab_case_name}"
44
45        websockets[snake_case_name] = {
46            "name": kebab_case_name,
47            "routeKey": route_key,
48            "type": EXTENSION_RUNTIME_MAP[extension.lower()]["language"],
49            "runtime": EXTENSION_RUNTIME_MAP[extension.lower()]["runtime"],
50            "zipName": f"{snake_case_name}.zip"
51        }
52
53    return websockets
54
55
56def dump_dict_to_file(filename: str, d: dict, folder="./"):
57    directory = f"{folder}/".replace("//", "/")
58    Path(directory).mkdir(parents=True, exist_ok=True)
59    with open(f"{directory}{filename}", "w") as f:
60        json.dump(d, f, indent=2)
61
62
63if __name__ == '__main__':
64    BASE_PATH = sys.argv[1]
65    print("🔵 [INFO] Generating websockets.json...")
66    mappings = _list_files_with_extensions(f"{BASE_PATH}/app/default-websockets")
67    dump_dict_to_file("websockets.json", make_websockets_json(mappings), f"{BASE_PATH}/tf.support.files/")
68    print("🟢 [SUCCESS] Done!")
BASE_PATH = None
EXTENSION_RUNTIME_MAP = {'py': {'language': 'python', 'runtime': 'python3.13'}}
def make_websockets_json(websocket_instances): 
20def make_websockets_json(websocket_instances):
21    """
22    Runs through each websocket and generates a JSON object with the websocket name.
23    Result example:
24    ```json
25    {
26          "connect": {
27            "name": "connect",
28            "type": "python",
29            "runtime": "python3.13",
30            "zipName": "connect.zip"
31          }
32    }
33    ```
34    """
35    websockets = {}
36    for websocket in websocket_instances:
37        filename = websocket[0]
38        extension = websocket[1]
39        snake_case_name = camelCase_to_snake_case(filename)
40        kebab_case_name = camelCase_to_kebab_case(filename)
41
42        route_key = kebab_case_name
43        if route_key in {"connect", "disconnect", "default"}:
44            route_key = f"${kebab_case_name}"
45
46        websockets[snake_case_name] = {
47            "name": kebab_case_name,
48            "routeKey": route_key,
49            "type": EXTENSION_RUNTIME_MAP[extension.lower()]["language"],
50            "runtime": EXTENSION_RUNTIME_MAP[extension.lower()]["runtime"],
51            "zipName": f"{snake_case_name}.zip"
52        }
53
54    return websockets

Runs through each websocket and generates a JSON object with the websocket name. Result example:

{
      "connect": {
        "name": "connect",
        "type": "python",
        "runtime": "python3.13",
        "zipName": "connect.zip"
      }
}
def dump_dict_to_file(filename: str, d: dict, folder='./'): 
57def dump_dict_to_file(filename: str, d: dict, folder="./"):
58    directory = f"{folder}/".replace("//", "/")
59    Path(directory).mkdir(parents=True, exist_ok=True)
60    with open(f"{directory}{filename}", "w") as f:
61        json.dump(d, f, indent=2)