Spaces:
Build error
Build error
File size: 2,588 Bytes
0827183 |
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 |
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
import json
from uuid import UUID, uuid4
from typing import List, Union
from msrest.serialization import Model
from botframework.streaming.payloads import ResponseMessageStream
from botframework.streaming.payloads.models import Serializable
class StreamingRequest:
GET = "GET"
POST = "POST"
PUT = "PUT"
DELETE = "DELETE"
def __init__(
self,
*,
verb: str = None,
path: str = None,
streams: List[ResponseMessageStream] = None,
):
self.verb = verb
self.path = path
self.streams = streams
@staticmethod
def create_request(
method: str, path: str = None, body: object = None
) -> "StreamingRequest":
if not method:
return None
request = StreamingRequest(
verb=method,
path=path,
)
if body:
request.add_stream(body)
return request
@staticmethod
def create_get(path: str = None, body: object = None) -> "StreamingRequest":
return StreamingRequest.create_request("GET", path, body)
@staticmethod
def create_post(path: str = None, body: object = None) -> "StreamingRequest":
return StreamingRequest.create_request("POST", path, body)
@staticmethod
def create_put(path: str = None, body: object = None) -> "StreamingRequest":
return StreamingRequest.create_request("PUT", path, body)
@staticmethod
def create_delete(path: str = None, body: object = None) -> "StreamingRequest":
return StreamingRequest.create_request("DELETE", path, body)
def set_body(self, body: Union[str, Serializable, Model, bytes]):
# TODO: verify if msrest.serialization.Model is necessary
if not body:
return
if isinstance(body, bytes):
pass
else:
if isinstance(body, Serializable):
body = body.to_json()
elif isinstance(body, Model):
body = json.dumps(body.as_dict())
body = body.encode("ascii")
self.add_stream(list(body))
def add_stream(self, content: object, stream_id: UUID = None):
if not content:
raise TypeError(
f"'content: {content.__class__.__name__}' argument can't be None"
)
if not self.streams:
self.streams = []
self.streams.append(
ResponseMessageStream(id=stream_id or uuid4(), content=content)
)
|