We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
fix: use StringIO instead of BytesIO with StreamWriter
Related to https://github.com/aws/aws-sam-cli/pull/5427
#### Which issue(s) does this change fix?
N/A


#### Why is this change necessary?
With recently changed the way to write into stream to fix some issues related to encoding (see PR above). That changed caused some issues with ECRUpload class, due to discrepancy of the `write` method between `BytesIO` vs `StringIO`. `BytesIO` instance accepts byte array where `StringIO` accepts string.

#### How does it address the issue?
This resolves issue by using `StringIO` stream instead of `BytesIO`. This PR also adds some typing for the inner `stream` instance of the `StreamWrapper` class to check for other usages in the code base.


#### What side effects does this change have?
Manually running `sam local` tests to validate the change.


#### Mandatory Checklist
**PRs will only be reviewed after checklist is complete**

- [ ] Add input/output [type hints](https://docs.python.org/3/library/typing.html) to new functions/methods
- [ ] Write design document if needed ([Do I need to write a design document?](https://github.com/aws/aws-sam-cli/blob/develop/DEVELOPMENT_GUIDE.md#design-document))
- [x] Write/update unit tests
- [ ] Write/update integration tests
- [ ] Write/update functional tests if needed
- [x] `make pr` passes
- [ ] `make update-reproducible-reqs` if dependencies were changed
- [ ] Write documentation

By submitting this pull request, I confirm that my contribution is made under the terms of the [Apache 2.0 license](https://www.apache.org/licenses/LICENSE-2.0).


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `schema/schema.py`
Content:
```
1 """Handles JSON schema generation logic"""
2 
3 
4 import importlib
5 import json
6 from dataclasses import dataclass
7 from enum import Enum
8 from typing import Any, Dict, List, Optional
9 
10 import click
11 
12 from samcli.cli.command import _SAM_CLI_COMMAND_PACKAGES
13 from samcli.lib.config.samconfig import SamConfig
14 
15 
16 class SchemaKeys(Enum):
17     SCHEMA_FILE_NAME = "schema/samcli.json"
18     SCHEMA_DRAFT = "http://json-schema.org/draft-04/schema"
19     TITLE = "AWS SAM CLI samconfig schema"
20     ENVIRONMENT_REGEX = "^.+$"
21 
22 
23 @dataclass()
24 class SamCliParameterSchema:
25     """Representation of a parameter in the SAM CLI.
26 
27     It includes relevant information for the JSON schema, such as name, data type,
28     and description, among others.
29     """
30 
31     name: str
32     type: str
33     description: str = ""
34     default: Optional[Any] = None
35     choices: Optional[Any] = None
36 
37     def to_schema(self) -> Dict[str, Any]:
38         """Return the JSON schema representation of the SAM CLI parameter."""
39         param = {}
40         param.update({"title": self.name, "type": self.type, "description": self.description})
41         if self.default:
42             param.update({"default": self.default})
43         if self.choices:
44             param.update({"enum": self.choices})
45         return param
46 
47 
48 @dataclass()
49 class SamCliCommandSchema:
50     """Representation of a command in the SAM CLI.
51 
52     It includes relevant information for the JSON schema, such as name, a description of the
53     command, and a list of all available parameters.
54     """
55 
56     name: str  # Full command name, with underscores (i.e. remote_invoke, local_start_lambda)
57     description: str
58     parameters: List[SamCliParameterSchema]
59 
60     def to_schema(self) -> dict:
61         """Return the JSON schema representation of the SAM CLI command."""
62         COMMANDS_TO_EXCLUDE = [  # TEMPORARY: for use only while generating piece-by-piece
63             "deploy",
64             "build",
65             "local",
66             "validate",
67             "package",
68             "init",
69             "delete",
70             "bootstrap",
71             "list",
72             "traces",
73             "sync",
74             "publish",
75             "pipeline",
76             "logs",
77             "remote",
78         ]
79         split_cmd_name = self.name.split("_")
80         formatted_cmd_name = " ".join(split_cmd_name)
81         exclude_params = split_cmd_name[0] in COMMANDS_TO_EXCLUDE
82         formatted_params_list = (
83             "* " + "\n* ".join([f"{param.name}:\n{param.description}" for param in self.parameters])
84             if not exclude_params
85             else ""
86         )
87         params_description = f"Available parameters for the {formatted_cmd_name} command:\n{formatted_params_list}"
88 
89         return {
90             self.name: {
91                 "title": f"{formatted_cmd_name.title()} command",
92                 "description": self.description or "",
93                 "properties": {
94                     "parameters": {
95                         "title": f"Parameters for the {formatted_cmd_name} command",
96                         "description": params_description,
97                         "type": "object",
98                         "properties": {param.name: param.to_schema() for param in self.parameters}
99                         if not exclude_params
100                         else {},
101                     },
102                 },
103                 "required": ["parameters"],
104             }
105         }
106 
107 
108 def clean_text(text: str) -> str:
109     """Clean up a string of text to be formatted for the JSON schema."""
110     if not text:
111         return ""
112     return text.replace("\b", "").strip("\n").strip()
113 
114 
115 def format_param(param: click.core.Option) -> SamCliParameterSchema:
116     """Format a click Option parameter to a SamCliParameter object.
117 
118     A parameter object should contain the following information that will be
119     necessary for including in the JSON schema:
120     * name - The name of the parameter
121     * help - The parameter's description (may vary between commands)
122     * type - The data type accepted by the parameter
123       * type.choices - If there are only a certain number of options allowed,
124                        a list of those allowed options
125     * default - The default option for that parameter
126     """
127     param_type = param.type.name.lower()
128     formatted_param_type = ""
129     # NOTE: Params do not have explicit "string" type; either "text" or "path".
130     #       All choice options are from a set of strings.
131     if param_type in ["text", "path", "choice", "filename", "directory"]:
132         formatted_param_type = "string"
133     elif param_type == "list":
134         formatted_param_type = "array"
135     else:
136         formatted_param_type = param_type or "string"
137 
138     formatted_param: SamCliParameterSchema = SamCliParameterSchema(
139         param.name or "", formatted_param_type, clean_text(param.help or "")
140     )
141 
142     if param.default:
143         formatted_param.default = list(param.default) if isinstance(param.default, tuple) else param.default
144 
145     if param.type.name == "choice" and isinstance(param.type, click.Choice):
146         formatted_param.choices = list(param.type.choices)
147 
148     return formatted_param
149 
150 
151 def get_params_from_command(cli) -> List[SamCliParameterSchema]:
152     """Given a CLI object, return a list of all parameters in that CLI, formatted as SamCliParameterSchema objects."""
153     return [format_param(param) for param in cli.params if param.name and isinstance(param, click.core.Option)]
154 
155 
156 def retrieve_command_structure(package_name: str) -> List[SamCliCommandSchema]:
157     """Given a SAM CLI package name, retrieve its structure.
158 
159     Such a structure is the list of all subcommands as `SamCliCommandSchema`, which includes
160     the command's name, description, and its parameters.
161 
162     Parameters
163     ----------
164     package_name: str
165         The name of the command package to retrieve.
166 
167     Returns
168     -------
169     List[SamCliCommandSchema]
170         A list of SamCliCommandSchema objects which represent either a command or a list of
171         subcommands within the package.
172     """
173     module = importlib.import_module(package_name)
174     command = []
175 
176     if isinstance(module.cli, click.core.Group):  # command has subcommands (e.g. local invoke)
177         for subcommand in module.cli.commands.values():
178             cmd_name = SamConfig.to_key([module.__name__.split(".")[-1], str(subcommand.name)])
179             command.append(
180                 SamCliCommandSchema(
181                     cmd_name,
182                     clean_text(subcommand.help or subcommand.short_help or ""),
183                     get_params_from_command(subcommand),
184                 )
185             )
186     else:
187         cmd_name = SamConfig.to_key([module.__name__.split(".")[-1]])
188         command.append(
189             SamCliCommandSchema(
190                 cmd_name,
191                 clean_text(module.cli.help or module.cli.short_help or ""),
192                 get_params_from_command(module.cli),
193             )
194         )
195     return command
196 
197 
198 def generate_schema() -> dict:
199     """Generate a JSON schema for all SAM CLI commands.
200 
201     Returns
202     -------
203     dict
204         A dictionary representation of the JSON schema.
205     """
206     schema: dict = {}
207     commands: List[SamCliCommandSchema] = []
208 
209     # Populate schema with relevant attributes
210     schema["$schema"] = SchemaKeys.SCHEMA_DRAFT.value
211     schema["title"] = SchemaKeys.TITLE.value
212     schema["type"] = "object"
213     schema["properties"] = {
214         # Version number required for samconfig files to be valid
215         "version": {"title": "Config version", "type": "number", "default": 0.1}
216     }
217     schema["required"] = ["version"]
218     schema["additionalProperties"] = False
219     # Iterate through packages for command and parameter information
220     for package_name in _SAM_CLI_COMMAND_PACKAGES:
221         commands.extend(retrieve_command_structure(package_name))
222     # Generate schema for each of the commands
223     schema["patternProperties"] = {SchemaKeys.ENVIRONMENT_REGEX.value: {"title": "Environment", "properties": {}}}
224     for command in commands:
225         schema["patternProperties"][SchemaKeys.ENVIRONMENT_REGEX.value]["properties"].update(command.to_schema())
226     return schema
227 
228 
229 def write_schema():
230     """Generate the SAM CLI JSON schema and write it to file."""
231     schema = generate_schema()
232     with open(SchemaKeys.SCHEMA_FILE_NAME.value, "w+", encoding="utf-8") as outfile:
233         json.dump(schema, outfile, indent=2)
234 
235 
236 if __name__ == "__main__":
237     write_schema()
238
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/schema/schema.py b/schema/schema.py
--- a/schema/schema.py
+++ b/schema/schema.py
@@ -32,14 +32,17 @@
     type: str
     description: str = ""
     default: Optional[Any] = None
+    items: Optional[str] = None
     choices: Optional[Any] = None
 
     def to_schema(self) -> Dict[str, Any]:
         """Return the JSON schema representation of the SAM CLI parameter."""
-        param = {}
+        param: Dict[str, Any] = {}
         param.update({"title": self.name, "type": self.type, "description": self.description})
         if self.default:
             param.update({"default": self.default})
+        if self.items:
+            param.update({"items": {"type": self.items}})
         if self.choices:
             param.update({"enum": self.choices})
         return param
@@ -136,7 +139,10 @@
         formatted_param_type = param_type or "string"
 
     formatted_param: SamCliParameterSchema = SamCliParameterSchema(
-        param.name or "", formatted_param_type, clean_text(param.help or "")
+        param.name or "",
+        formatted_param_type,
+        clean_text(param.help or ""),
+        items="string" if formatted_param_type == "array" else None,
     )
 
     if param.default:
@@ -150,7 +156,15 @@
 
 def get_params_from_command(cli) -> List[SamCliParameterSchema]:
     """Given a CLI object, return a list of all parameters in that CLI, formatted as SamCliParameterSchema objects."""
-    return [format_param(param) for param in cli.params if param.name and isinstance(param, click.core.Option)]
+    params_to_exclude = [
+        "config_env",  # shouldn't allow different environment from where the config is being read from
+        "config_file",  # shouldn't allow reading another file within current file
+    ]
+    return [
+        format_param(param)
+        for param in cli.params
+        if param.name and isinstance(param, click.core.Option) and param.name not in params_to_exclude
+    ]
 
 
 def retrieve_command_structure(package_name: str) -> List[SamCliCommandSchema]:

{"golden_diff": "diff --git a/schema/schema.py b/schema/schema.py\n--- a/schema/schema.py\n+++ b/schema/schema.py\n@@ -32,14 +32,17 @@\n     type: str\n     description: str = \"\"\n     default: Optional[Any] = None\n+    items: Optional[str] = None\n     choices: Optional[Any] = None\n \n     def to_schema(self) -> Dict[str, Any]:\n         \"\"\"Return the JSON schema representation of the SAM CLI parameter.\"\"\"\n-        param = {}\n+        param: Dict[str, Any] = {}\n         param.update({\"title\": self.name, \"type\": self.type, \"description\": self.description})\n         if self.default:\n             param.update({\"default\": self.default})\n+        if self.items:\n+            param.update({\"items\": {\"type\": self.items}})\n         if self.choices:\n             param.update({\"enum\": self.choices})\n         return param\n@@ -136,7 +139,10 @@\n         formatted_param_type = param_type or \"string\"\n \n     formatted_param: SamCliParameterSchema = SamCliParameterSchema(\n-        param.name or \"\", formatted_param_type, clean_text(param.help or \"\")\n+        param.name or \"\",\n+        formatted_param_type,\n+        clean_text(param.help or \"\"),\n+        items=\"string\" if formatted_param_type == \"array\" else None,\n     )\n \n     if param.default:\n@@ -150,7 +156,15 @@\n \n def get_params_from_command(cli) -> List[SamCliParameterSchema]:\n     \"\"\"Given a CLI object, return a list of all parameters in that CLI, formatted as SamCliParameterSchema objects.\"\"\"\n-    return [format_param(param) for param in cli.params if param.name and isinstance(param, click.core.Option)]\n+    params_to_exclude = [\n+        \"config_env\",  # shouldn't allow different environment from where the config is being read from\n+        \"config_file\",  # shouldn't allow reading another file within current file\n+    ]\n+    return [\n+        format_param(param)\n+        for param in cli.params\n+        if param.name and isinstance(param, click.core.Option) and param.name not in params_to_exclude\n+    ]\n \n \n def retrieve_command_structure(package_name: str) -> List[SamCliCommandSchema]:\n", "issue": "fix: use StringIO instead of BytesIO with StreamWriter\nRelated to https://github.com/aws/aws-sam-cli/pull/5427\r\n#### Which issue(s) does this change fix?\r\nN/A\r\n\r\n\r\n#### Why is this change necessary?\r\nWith recently changed the way to write into stream to fix some issues related to encoding (see PR above). That changed caused some issues with ECRUpload class, due to discrepancy of the `write` method between `BytesIO` vs `StringIO`. `BytesIO` instance accepts byte array where `StringIO` accepts string.\r\n\r\n#### How does it address the issue?\r\nThis resolves issue by using `StringIO` stream instead of `BytesIO`. This PR also adds some typing for the inner `stream` instance of the `StreamWrapper` class to check for other usages in the code base.\r\n\r\n\r\n#### What side effects does this change have?\r\nManually running `sam local` tests to validate the change.\r\n\r\n\r\n#### Mandatory Checklist\r\n**PRs will only be reviewed after checklist is complete**\r\n\r\n- [ ] Add input/output [type hints](https://docs.python.org/3/library/typing.html) to new functions/methods\r\n- [ ] Write design document if needed ([Do I need to write a design document?](https://github.com/aws/aws-sam-cli/blob/develop/DEVELOPMENT_GUIDE.md#design-document))\r\n- [x] Write/update unit tests\r\n- [ ] Write/update integration tests\r\n- [ ] Write/update functional tests if needed\r\n- [x] `make pr` passes\r\n- [ ] `make update-reproducible-reqs` if dependencies were changed\r\n- [ ] Write documentation\r\n\r\nBy submitting this pull request, I confirm that my contribution is made under the terms of the [Apache 2.0 license](https://www.apache.org/licenses/LICENSE-2.0).\r\n\n", "before_files": [{"content": "\"\"\"Handles JSON schema generation logic\"\"\"\n\n\nimport importlib\nimport json\nfrom dataclasses import dataclass\nfrom enum import Enum\nfrom typing import Any, Dict, List, Optional\n\nimport click\n\nfrom samcli.cli.command import _SAM_CLI_COMMAND_PACKAGES\nfrom samcli.lib.config.samconfig import SamConfig\n\n\nclass SchemaKeys(Enum):\n    SCHEMA_FILE_NAME = \"schema/samcli.json\"\n    SCHEMA_DRAFT = \"http://json-schema.org/draft-04/schema\"\n    TITLE = \"AWS SAM CLI samconfig schema\"\n    ENVIRONMENT_REGEX = \"^.+$\"\n\n\n@dataclass()\nclass SamCliParameterSchema:\n    \"\"\"Representation of a parameter in the SAM CLI.\n\n    It includes relevant information for the JSON schema, such as name, data type,\n    and description, among others.\n    \"\"\"\n\n    name: str\n    type: str\n    description: str = \"\"\n    default: Optional[Any] = None\n    choices: Optional[Any] = None\n\n    def to_schema(self) -> Dict[str, Any]:\n        \"\"\"Return the JSON schema representation of the SAM CLI parameter.\"\"\"\n        param = {}\n        param.update({\"title\": self.name, \"type\": self.type, \"description\": self.description})\n        if self.default:\n            param.update({\"default\": self.default})\n        if self.choices:\n            param.update({\"enum\": self.choices})\n        return param\n\n\n@dataclass()\nclass SamCliCommandSchema:\n    \"\"\"Representation of a command in the SAM CLI.\n\n    It includes relevant information for the JSON schema, such as name, a description of the\n    command, and a list of all available parameters.\n    \"\"\"\n\n    name: str  # Full command name, with underscores (i.e. remote_invoke, local_start_lambda)\n    description: str\n    parameters: List[SamCliParameterSchema]\n\n    def to_schema(self) -> dict:\n        \"\"\"Return the JSON schema representation of the SAM CLI command.\"\"\"\n        COMMANDS_TO_EXCLUDE = [  # TEMPORARY: for use only while generating piece-by-piece\n            \"deploy\",\n            \"build\",\n            \"local\",\n            \"validate\",\n            \"package\",\n            \"init\",\n            \"delete\",\n            \"bootstrap\",\n            \"list\",\n            \"traces\",\n            \"sync\",\n            \"publish\",\n            \"pipeline\",\n            \"logs\",\n            \"remote\",\n        ]\n        split_cmd_name = self.name.split(\"_\")\n        formatted_cmd_name = \" \".join(split_cmd_name)\n        exclude_params = split_cmd_name[0] in COMMANDS_TO_EXCLUDE\n        formatted_params_list = (\n            \"* \" + \"\\n* \".join([f\"{param.name}:\\n{param.description}\" for param in self.parameters])\n            if not exclude_params\n            else \"\"\n        )\n        params_description = f\"Available parameters for the {formatted_cmd_name} command:\\n{formatted_params_list}\"\n\n        return {\n            self.name: {\n                \"title\": f\"{formatted_cmd_name.title()} command\",\n                \"description\": self.description or \"\",\n                \"properties\": {\n                    \"parameters\": {\n                        \"title\": f\"Parameters for the {formatted_cmd_name} command\",\n                        \"description\": params_description,\n                        \"type\": \"object\",\n                        \"properties\": {param.name: param.to_schema() for param in self.parameters}\n                        if not exclude_params\n                        else {},\n                    },\n                },\n                \"required\": [\"parameters\"],\n            }\n        }\n\n\ndef clean_text(text: str) -> str:\n    \"\"\"Clean up a string of text to be formatted for the JSON schema.\"\"\"\n    if not text:\n        return \"\"\n    return text.replace(\"\\b\", \"\").strip(\"\\n\").strip()\n\n\ndef format_param(param: click.core.Option) -> SamCliParameterSchema:\n    \"\"\"Format a click Option parameter to a SamCliParameter object.\n\n    A parameter object should contain the following information that will be\n    necessary for including in the JSON schema:\n    * name - The name of the parameter\n    * help - The parameter's description (may vary between commands)\n    * type - The data type accepted by the parameter\n      * type.choices - If there are only a certain number of options allowed,\n                       a list of those allowed options\n    * default - The default option for that parameter\n    \"\"\"\n    param_type = param.type.name.lower()\n    formatted_param_type = \"\"\n    # NOTE: Params do not have explicit \"string\" type; either \"text\" or \"path\".\n    #       All choice options are from a set of strings.\n    if param_type in [\"text\", \"path\", \"choice\", \"filename\", \"directory\"]:\n        formatted_param_type = \"string\"\n    elif param_type == \"list\":\n        formatted_param_type = \"array\"\n    else:\n        formatted_param_type = param_type or \"string\"\n\n    formatted_param: SamCliParameterSchema = SamCliParameterSchema(\n        param.name or \"\", formatted_param_type, clean_text(param.help or \"\")\n    )\n\n    if param.default:\n        formatted_param.default = list(param.default) if isinstance(param.default, tuple) else param.default\n\n    if param.type.name == \"choice\" and isinstance(param.type, click.Choice):\n        formatted_param.choices = list(param.type.choices)\n\n    return formatted_param\n\n\ndef get_params_from_command(cli) -> List[SamCliParameterSchema]:\n    \"\"\"Given a CLI object, return a list of all parameters in that CLI, formatted as SamCliParameterSchema objects.\"\"\"\n    return [format_param(param) for param in cli.params if param.name and isinstance(param, click.core.Option)]\n\n\ndef retrieve_command_structure(package_name: str) -> List[SamCliCommandSchema]:\n    \"\"\"Given a SAM CLI package name, retrieve its structure.\n\n    Such a structure is the list of all subcommands as `SamCliCommandSchema`, which includes\n    the command's name, description, and its parameters.\n\n    Parameters\n    ----------\n    package_name: str\n        The name of the command package to retrieve.\n\n    Returns\n    -------\n    List[SamCliCommandSchema]\n        A list of SamCliCommandSchema objects which represent either a command or a list of\n        subcommands within the package.\n    \"\"\"\n    module = importlib.import_module(package_name)\n    command = []\n\n    if isinstance(module.cli, click.core.Group):  # command has subcommands (e.g. local invoke)\n        for subcommand in module.cli.commands.values():\n            cmd_name = SamConfig.to_key([module.__name__.split(\".\")[-1], str(subcommand.name)])\n            command.append(\n                SamCliCommandSchema(\n                    cmd_name,\n                    clean_text(subcommand.help or subcommand.short_help or \"\"),\n                    get_params_from_command(subcommand),\n                )\n            )\n    else:\n        cmd_name = SamConfig.to_key([module.__name__.split(\".\")[-1]])\n        command.append(\n            SamCliCommandSchema(\n                cmd_name,\n                clean_text(module.cli.help or module.cli.short_help or \"\"),\n                get_params_from_command(module.cli),\n            )\n        )\n    return command\n\n\ndef generate_schema() -> dict:\n    \"\"\"Generate a JSON schema for all SAM CLI commands.\n\n    Returns\n    -------\n    dict\n        A dictionary representation of the JSON schema.\n    \"\"\"\n    schema: dict = {}\n    commands: List[SamCliCommandSchema] = []\n\n    # Populate schema with relevant attributes\n    schema[\"$schema\"] = SchemaKeys.SCHEMA_DRAFT.value\n    schema[\"title\"] = SchemaKeys.TITLE.value\n    schema[\"type\"] = \"object\"\n    schema[\"properties\"] = {\n        # Version number required for samconfig files to be valid\n        \"version\": {\"title\": \"Config version\", \"type\": \"number\", \"default\": 0.1}\n    }\n    schema[\"required\"] = [\"version\"]\n    schema[\"additionalProperties\"] = False\n    # Iterate through packages for command and parameter information\n    for package_name in _SAM_CLI_COMMAND_PACKAGES:\n        commands.extend(retrieve_command_structure(package_name))\n    # Generate schema for each of the commands\n    schema[\"patternProperties\"] = {SchemaKeys.ENVIRONMENT_REGEX.value: {\"title\": \"Environment\", \"properties\": {}}}\n    for command in commands:\n        schema[\"patternProperties\"][SchemaKeys.ENVIRONMENT_REGEX.value][\"properties\"].update(command.to_schema())\n    return schema\n\n\ndef write_schema():\n    \"\"\"Generate the SAM CLI JSON schema and write it to file.\"\"\"\n    schema = generate_schema()\n    with open(SchemaKeys.SCHEMA_FILE_NAME.value, \"w+\", encoding=\"utf-8\") as outfile:\n        json.dump(schema, outfile, indent=2)\n\n\nif __name__ == \"__main__\":\n    write_schema()\n", "path": "schema/schema.py"}], "after_files": [{"content": "\"\"\"Handles JSON schema generation logic\"\"\"\n\n\nimport importlib\nimport json\nfrom dataclasses import dataclass\nfrom enum import Enum\nfrom typing import Any, Dict, List, Optional\n\nimport click\n\nfrom samcli.cli.command import _SAM_CLI_COMMAND_PACKAGES\nfrom samcli.lib.config.samconfig import SamConfig\n\n\nclass SchemaKeys(Enum):\n    SCHEMA_FILE_NAME = \"schema/samcli.json\"\n    SCHEMA_DRAFT = \"http://json-schema.org/draft-04/schema\"\n    TITLE = \"AWS SAM CLI samconfig schema\"\n    ENVIRONMENT_REGEX = \"^.+$\"\n\n\n@dataclass()\nclass SamCliParameterSchema:\n    \"\"\"Representation of a parameter in the SAM CLI.\n\n    It includes relevant information for the JSON schema, such as name, data type,\n    and description, among others.\n    \"\"\"\n\n    name: str\n    type: str\n    description: str = \"\"\n    default: Optional[Any] = None\n    items: Optional[str] = None\n    choices: Optional[Any] = None\n\n    def to_schema(self) -> Dict[str, Any]:\n        \"\"\"Return the JSON schema representation of the SAM CLI parameter.\"\"\"\n        param: Dict[str, Any] = {}\n        param.update({\"title\": self.name, \"type\": self.type, \"description\": self.description})\n        if self.default:\n            param.update({\"default\": self.default})\n        if self.items:\n            param.update({\"items\": {\"type\": self.items}})\n        if self.choices:\n            param.update({\"enum\": self.choices})\n        return param\n\n\n@dataclass()\nclass SamCliCommandSchema:\n    \"\"\"Representation of a command in the SAM CLI.\n\n    It includes relevant information for the JSON schema, such as name, a description of the\n    command, and a list of all available parameters.\n    \"\"\"\n\n    name: str  # Full command name, with underscores (i.e. remote_invoke, local_start_lambda)\n    description: str\n    parameters: List[SamCliParameterSchema]\n\n    def to_schema(self) -> dict:\n        \"\"\"Return the JSON schema representation of the SAM CLI command.\"\"\"\n        COMMANDS_TO_EXCLUDE = [  # TEMPORARY: for use only while generating piece-by-piece\n            \"deploy\",\n            \"build\",\n            \"local\",\n            \"validate\",\n            \"package\",\n            \"init\",\n            \"delete\",\n            \"bootstrap\",\n            \"list\",\n            \"traces\",\n            \"sync\",\n            \"publish\",\n            \"pipeline\",\n            \"logs\",\n            \"remote\",\n        ]\n        split_cmd_name = self.name.split(\"_\")\n        formatted_cmd_name = \" \".join(split_cmd_name)\n        exclude_params = split_cmd_name[0] in COMMANDS_TO_EXCLUDE\n        formatted_params_list = (\n            \"* \" + \"\\n* \".join([f\"{param.name}:\\n{param.description}\" for param in self.parameters])\n            if not exclude_params\n            else \"\"\n        )\n        params_description = f\"Available parameters for the {formatted_cmd_name} command:\\n{formatted_params_list}\"\n\n        return {\n            self.name: {\n                \"title\": f\"{formatted_cmd_name.title()} command\",\n                \"description\": self.description or \"\",\n                \"properties\": {\n                    \"parameters\": {\n                        \"title\": f\"Parameters for the {formatted_cmd_name} command\",\n                        \"description\": params_description,\n                        \"type\": \"object\",\n                        \"properties\": {param.name: param.to_schema() for param in self.parameters}\n                        if not exclude_params\n                        else {},\n                    },\n                },\n                \"required\": [\"parameters\"],\n            }\n        }\n\n\ndef clean_text(text: str) -> str:\n    \"\"\"Clean up a string of text to be formatted for the JSON schema.\"\"\"\n    if not text:\n        return \"\"\n    return text.replace(\"\\b\", \"\").strip(\"\\n\").strip()\n\n\ndef format_param(param: click.core.Option) -> SamCliParameterSchema:\n    \"\"\"Format a click Option parameter to a SamCliParameter object.\n\n    A parameter object should contain the following information that will be\n    necessary for including in the JSON schema:\n    * name - The name of the parameter\n    * help - The parameter's description (may vary between commands)\n    * type - The data type accepted by the parameter\n      * type.choices - If there are only a certain number of options allowed,\n                       a list of those allowed options\n    * default - The default option for that parameter\n    \"\"\"\n    param_type = param.type.name.lower()\n    formatted_param_type = \"\"\n    # NOTE: Params do not have explicit \"string\" type; either \"text\" or \"path\".\n    #       All choice options are from a set of strings.\n    if param_type in [\"text\", \"path\", \"choice\", \"filename\", \"directory\"]:\n        formatted_param_type = \"string\"\n    elif param_type == \"list\":\n        formatted_param_type = \"array\"\n    else:\n        formatted_param_type = param_type or \"string\"\n\n    formatted_param: SamCliParameterSchema = SamCliParameterSchema(\n        param.name or \"\",\n        formatted_param_type,\n        clean_text(param.help or \"\"),\n        items=\"string\" if formatted_param_type == \"array\" else None,\n    )\n\n    if param.default:\n        formatted_param.default = list(param.default) if isinstance(param.default, tuple) else param.default\n\n    if param.type.name == \"choice\" and isinstance(param.type, click.Choice):\n        formatted_param.choices = list(param.type.choices)\n\n    return formatted_param\n\n\ndef get_params_from_command(cli) -> List[SamCliParameterSchema]:\n    \"\"\"Given a CLI object, return a list of all parameters in that CLI, formatted as SamCliParameterSchema objects.\"\"\"\n    params_to_exclude = [\n        \"config_env\",  # shouldn't allow different environment from where the config is being read from\n        \"config_file\",  # shouldn't allow reading another file within current file\n    ]\n    return [\n        format_param(param)\n        for param in cli.params\n        if param.name and isinstance(param, click.core.Option) and param.name not in params_to_exclude\n    ]\n\n\ndef retrieve_command_structure(package_name: str) -> List[SamCliCommandSchema]:\n    \"\"\"Given a SAM CLI package name, retrieve its structure.\n\n    Such a structure is the list of all subcommands as `SamCliCommandSchema`, which includes\n    the command's name, description, and its parameters.\n\n    Parameters\n    ----------\n    package_name: str\n        The name of the command package to retrieve.\n\n    Returns\n    -------\n    List[SamCliCommandSchema]\n        A list of SamCliCommandSchema objects which represent either a command or a list of\n        subcommands within the package.\n    \"\"\"\n    module = importlib.import_module(package_name)\n    command = []\n\n    if isinstance(module.cli, click.core.Group):  # command has subcommands (e.g. local invoke)\n        for subcommand in module.cli.commands.values():\n            cmd_name = SamConfig.to_key([module.__name__.split(\".\")[-1], str(subcommand.name)])\n            command.append(\n                SamCliCommandSchema(\n                    cmd_name,\n                    clean_text(subcommand.help or subcommand.short_help or \"\"),\n                    get_params_from_command(subcommand),\n                )\n            )\n    else:\n        cmd_name = SamConfig.to_key([module.__name__.split(\".\")[-1]])\n        command.append(\n            SamCliCommandSchema(\n                cmd_name,\n                clean_text(module.cli.help or module.cli.short_help or \"\"),\n                get_params_from_command(module.cli),\n            )\n        )\n    return command\n\n\ndef generate_schema() -> dict:\n    \"\"\"Generate a JSON schema for all SAM CLI commands.\n\n    Returns\n    -------\n    dict\n        A dictionary representation of the JSON schema.\n    \"\"\"\n    schema: dict = {}\n    commands: List[SamCliCommandSchema] = []\n\n    # Populate schema with relevant attributes\n    schema[\"$schema\"] = SchemaKeys.SCHEMA_DRAFT.value\n    schema[\"title\"] = SchemaKeys.TITLE.value\n    schema[\"type\"] = \"object\"\n    schema[\"properties\"] = {\n        # Version number required for samconfig files to be valid\n        \"version\": {\"title\": \"Config version\", \"type\": \"number\", \"default\": 0.1}\n    }\n    schema[\"required\"] = [\"version\"]\n    schema[\"additionalProperties\"] = False\n    # Iterate through packages for command and parameter information\n    for package_name in _SAM_CLI_COMMAND_PACKAGES:\n        commands.extend(retrieve_command_structure(package_name))\n    # Generate schema for each of the commands\n    schema[\"patternProperties\"] = {SchemaKeys.ENVIRONMENT_REGEX.value: {\"title\": \"Environment\", \"properties\": {}}}\n    for command in commands:\n        schema[\"patternProperties\"][SchemaKeys.ENVIRONMENT_REGEX.value][\"properties\"].update(command.to_schema())\n    return schema\n\n\ndef write_schema():\n    \"\"\"Generate the SAM CLI JSON schema and write it to file.\"\"\"\n    schema = generate_schema()\n    with open(SchemaKeys.SCHEMA_FILE_NAME.value, \"w+\", encoding=\"utf-8\") as outfile:\n        json.dump(schema, outfile, indent=2)\n\n\nif __name__ == \"__main__\":\n    write_schema()\n", "path": "schema/schema.py"}]}