1import json 2 3from moto.core.responses import BaseResponse 4from .models import mediaconnect_backends 5 6try: 7 from urllib import unquote 8except ImportError: 9 from urllib.parse import unquote 10 11 12class MediaConnectResponse(BaseResponse): 13 SERVICE_NAME = "mediaconnect" 14 15 @property 16 def mediaconnect_backend(self): 17 return mediaconnect_backends[self.region] 18 19 def create_flow(self): 20 availability_zone = self._get_param("availabilityZone") 21 entitlements = self._get_param("entitlements") 22 name = self._get_param("name") 23 outputs = self._get_param("outputs") 24 source = self._get_param("source") 25 source_failover_config = self._get_param("sourceFailoverConfig") 26 sources = self._get_param("sources") 27 vpc_interfaces = self._get_param("vpcInterfaces") 28 flow = self.mediaconnect_backend.create_flow( 29 availability_zone=availability_zone, 30 entitlements=entitlements, 31 name=name, 32 outputs=outputs, 33 source=source, 34 source_failover_config=source_failover_config, 35 sources=sources, 36 vpc_interfaces=vpc_interfaces, 37 ) 38 return json.dumps(dict(flow=flow.to_dict())) 39 40 def list_flows(self): 41 max_results = self._get_int_param("maxResults") 42 next_token = self._get_param("nextToken") 43 flows, next_token = self.mediaconnect_backend.list_flows( 44 max_results=max_results, next_token=next_token, 45 ) 46 return json.dumps(dict(flows=flows, nextToken=next_token)) 47 48 def describe_flow(self): 49 flow_arn = unquote(self._get_param("flowArn")) 50 flow, messages = self.mediaconnect_backend.describe_flow(flow_arn=flow_arn,) 51 return json.dumps(dict(flow=flow, messages=messages)) 52 53 def delete_flow(self): 54 flow_arn = unquote(self._get_param("flowArn")) 55 flow_arn, status = self.mediaconnect_backend.delete_flow(flow_arn=flow_arn,) 56 return json.dumps(dict(flowArn=flow_arn, status=status)) 57 58 def start_flow(self): 59 flow_arn = unquote(self._get_param("flowArn")) 60 flow_arn, status = self.mediaconnect_backend.start_flow(flow_arn=flow_arn,) 61 return json.dumps(dict(flowArn=flow_arn, status=status)) 62 63 def stop_flow(self): 64 flow_arn = unquote(self._get_param("flowArn")) 65 flow_arn, status = self.mediaconnect_backend.stop_flow(flow_arn=flow_arn,) 66 return json.dumps(dict(flowArn=flow_arn, status=status)) 67 68 def tag_resource(self): 69 resource_arn = unquote(self._get_param("resourceArn")) 70 tags = self._get_param("tags") 71 self.mediaconnect_backend.tag_resource( 72 resource_arn=resource_arn, tags=tags, 73 ) 74 return json.dumps(dict()) 75 76 def list_tags_for_resource(self): 77 resource_arn = unquote(self._get_param("resourceArn")) 78 tags = self.mediaconnect_backend.list_tags_for_resource( 79 resource_arn=resource_arn, 80 ) 81 return json.dumps(dict(tags=tags)) 82 83 def add_flow_vpc_interfaces(self): 84 flow_arn = unquote(self._get_param("flowArn")) 85 vpc_interfaces = self._get_param("vpcInterfaces") 86 flow_arn, vpc_interfaces = self.mediaconnect_backend.add_flow_vpc_interfaces( 87 flow_arn=flow_arn, vpc_interfaces=vpc_interfaces 88 ) 89 return json.dumps(dict(flow_arn=flow_arn, vpc_interfaces=vpc_interfaces)) 90 91 def remove_flow_vpc_interface(self): 92 flow_arn = unquote(self._get_param("flowArn")) 93 vpc_interface_name = unquote(self._get_param("vpcInterfaceName")) 94 ( 95 flow_arn, 96 vpc_interface_name, 97 ) = self.mediaconnect_backend.remove_flow_vpc_interface( 98 flow_arn=flow_arn, vpc_interface_name=vpc_interface_name 99 ) 100 return json.dumps( 101 dict(flow_arn=flow_arn, vpc_interface_name=vpc_interface_name) 102 ) 103 104 def add_flow_outputs(self): 105 flow_arn = unquote(self._get_param("flowArn")) 106 outputs = self._get_param("outputs") 107 flow_arn, outputs = self.mediaconnect_backend.add_flow_outputs( 108 flow_arn=flow_arn, outputs=outputs 109 ) 110 return json.dumps(dict(flow_arn=flow_arn, outputs=outputs)) 111 112 def remove_flow_output(self): 113 flow_arn = unquote(self._get_param("flowArn")) 114 output_name = unquote(self._get_param("outputArn")) 115 flow_arn, output_name = self.mediaconnect_backend.remove_flow_output( 116 flow_arn=flow_arn, output_name=output_name 117 ) 118 return json.dumps(dict(flow_arn=flow_arn, output_name=output_name)) 119 120 # add methods from here 121