in src/handlers/ruleset_handler.py [0:0]
def release_ruleset(self, event: RulesetReleasePostModel):
customer = event.customer or SYSTEM_CUSTOMER
client = self.license_manager_service.cl
if not isinstance(client, LMClientAfter3p0):
raise ResponseFactory(HTTPStatus.NOT_IMPLEMENTED).message(
'The linked License Manager does not support this operation'
).exc()
if event.is_all_versions:
rulesets = list(self.ruleset_service.iter_standard(
customer=customer,
name=event.name
))
elif event.version:
ruleset = self.ruleset_service.get_standard(
customer=customer,
name=event.name,
version=Version(event.version).to_str()
)
if not ruleset:
raise ResponseFactory(HTTPStatus.NOT_FOUND).message(
f'Ruleset with name {event.name} and '
f'version {event.version} not found'
).exc()
rulesets = [ruleset]
else:
ruleset = self.ruleset_service.get_latest(
customer=customer,
name=event.name
)
if not ruleset:
raise ResponseFactory(HTTPStatus.NOT_FOUND).message(
f'No rulesets with name {event.name} exist'
).exc()
rulesets = [ruleset]
responses = []
for rs in rulesets:
result = client.post_ruleset(
name=rs.name,
version=rs.version,
cloud=rs.cloud,
description=event.description,
display_name=event.display_name,
download_url=self.ruleset_service.download_url(rs),
rules=rs.rules
)
if not result:
code = HTTPStatus.SERVICE_UNAVAILABLE
released = False
message = 'Problem with the License Manager occurred'
else:
code, message = result
released = code is HTTPStatus.CREATED
responses.append(dict(
name=rs.name,
version=rs.version,
released=released,
message=message,
code=code
))
if all([item['released'] for item in responses]):
result_code = HTTPStatus.CREATED
else:
result_code = HTTPStatus.MULTI_STATUS
return build_response(code=result_code, content=responses)