1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
import argparse
import logging
from pathlib import Path
import os
from jinja2 import Environment, FileSystemLoader
from subprocess import check_call
from typing import Any
import json
_LOGGER = logging.getLogger(__name__)
_TEMPLATE = Path(__file__).resolve().parent / "template"
_TEMPLATE_TESTS = Path(__file__).resolve().parent / "template_tests"
_TEMPLATE_SAMPLES = Path(__file__).resolve().parent / "template_samples"
_TEMPLATE_CI = Path(__file__).resolve().parent / "template_ci"
_CONFIG_FILE = Path(__file__).resolve().parent / "../../swagger_to_sdk_config_dpg.json"
def check_parameters(
output_folder: str,
) -> None:
# check output_folder exists or not. If not, create it.
output = Path(output_folder)
if not os.path.exists(output):
_LOGGER.info(f'{output} does not exist and try to create it')
os.makedirs(output)
_LOGGER.info(f'{output} is created')
def generate_ci(template_path: Path, folder_path: Path, package_name: str) -> None:
ci = Path(folder_path, "ci.yml")
service_name = folder_path.name
safe_name = package_name.replace("-", "")
if not ci.exists():
env = Environment(loader=FileSystemLoader(template_path), keep_trailing_newline=True)
template = env.get_template('ci.yml')
content = template.render(package_name=package_name, service_name=service_name, safe_name=safe_name)
else:
with open(ci, "r") as file_in:
content = file_in.readlines()
for line in content:
if package_name in line:
return
content.append(f' - name: {package_name}\n')
content.append(f' safeName: {safe_name}\n')
with open(ci, "w") as file_out:
file_out.writelines(content)
def generate_test_sample(template_path: Path, target_path: Path, **kwargs: Any) -> None:
if not os.path.exists(target_path):
os.makedirs(target_path)
env = Environment(loader=FileSystemLoader(template_path), keep_trailing_newline=True)
for template_name in env.list_templates():
_LOGGER.info(f"generate file: {template_name}")
template = env.get_template(template_name)
result = template.render(**kwargs)
with open(target_path / template_name, "w") as fd:
fd.write(result)
def generate_swagger_readme(work_path: str, env: Environment, **kwargs: Any) -> Path:
_LOGGER.info("Building swagger readme")
# check path exists
swagger_path = Path(work_path) / Path('swagger')
if not os.path.exists(swagger_path):
os.makedirs(swagger_path)
# render file
template = env.get_template('README.md')
input_file = kwargs.pop("input_file", "").split(",")
result = template.render(input_file=input_file, **kwargs)
swagger_readme = swagger_path / Path('README.md')
with open(swagger_readme, 'w') as fd:
fd.write(result)
return swagger_readme
def generate_toml_file(target_path: Path) -> None:
with open(target_path / "sdk_packaging.toml", "w") as file_out:
file_out.write("[packaging]\nauto_update = false\n")
def get_autorest_version() -> str:
with open(_CONFIG_FILE, 'r') as file_in:
config = json.load(file_in)
autorest_use = " ".join(["--use=" + item for item in config["meta"]["autorest_options"]["use"]])
return "--version={} {}".format(config["meta"]["autorest_options"]["version"], autorest_use)
def build_package(**kwargs) -> None:
# prepare template render parameters
output_folder = kwargs.get("output_folder")
package_name = kwargs.get("package_name")
namespace = package_name.replace('-', '.')
kwargs['namespace'] = namespace
kwargs['test_prefix'] = package_name.split('-')[-1]
_LOGGER.info("Build start: %s", package_name)
check_parameters(output_folder)
# generate ci
generate_ci(_TEMPLATE_CI, Path(output_folder).parent, package_name)
# generate swagger readme
env = Environment(loader=FileSystemLoader(_TEMPLATE), keep_trailing_newline=True)
swagger_readme = generate_swagger_readme(output_folder, env, **kwargs)
# generate code with autorest and swagger readme
autorest_cmd = f'autorest {swagger_readme} {get_autorest_version()} '
_LOGGER.info(f"generate SDK code with autorest: {autorest_cmd}")
check_call(autorest_cmd, shell=True)
# generate test framework
work_path = Path(output_folder)
generate_test_sample(_TEMPLATE_TESTS, work_path / Path('tests'), **kwargs)
# generate sample framework
generate_test_sample(_TEMPLATE_SAMPLES, work_path / Path('samples'), **kwargs)
# generate .toml file to avoid udpate_pr in CI
generate_toml_file(work_path)
_LOGGER.info("Build complete: %s", package_name)
def validate_params(**kwargs):
if not kwargs.get("security_scope") and not kwargs.get("security_header_name"):
raise Exception('At least one of "security-scope" and "security-header-name" is needed')
def main(**kwargs):
build_package(**kwargs)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="build package for Azure SDK of data-plane for Python",
formatter_class=argparse.RawTextHelpFormatter,
)
parser.add_argument(
"--output-folder", "-o",
dest="output_folder",
required=True,
help="absolute path where generated SDK package will be put"
)
parser.add_argument("--debug", dest="debug", action="store_true", help="Verbosity in DEBUG mode")
parser.add_argument(
"--input-file", "-f",
dest="input_file",
required=True,
help="absolute path of swagger input file. For example: `https://raw.githubusercontent.com/Azure/azure-rest-api-specs/main/specification/webpubsub/data-plane/WebPubSub/stable/2021-10-01/webpubsub.json`"
" or `D:\\azure-rest-api-specs\\specification\\webpubsub\\data-plane\\WebPubSub\\stable\\2021-10-01\\webpubsub.json`",
)
parser.add_argument(
"--security-scope", "-c",
dest="security_scope",
required=False,
help="If authentication is AADToken, this param is necessary",
)
parser.add_argument(
"--security-header-name",
dest="security_header_name",
required=False,
help="If authentication is api key, this param is necessary",
)
parser.add_argument(
"--package-name", "-p",
dest="package_name",
required=True,
help="package name. For example: azure-messaging-webpubsub",
)
parser.add_argument(
"--package-pprint-name", "-n",
dest="package_pprint_name",
required=True,
help="Print name of the package. For example: Azure Web PubSub Service",
)
parser.add_argument(
"--client-name", "-t",
dest="client_name",
required=True,
help="client name. For example: WebPubSubServiceClient",
)
args = parser.parse_args()
main_logger = logging.getLogger()
logging.basicConfig()
main_logger.setLevel(logging.INFO)
parameters = vars(args)
validate_params(**parameters)
main(**parameters)
|
