I/O

plateforme.core.api.datastructures

This module provides utilities for managing data structures within the Plateforme framework's API using FastAPI and Starlette features.

URL

URL(
    url: str = "",
    scope: Scope | None = None,
    **components: Any,
)

Source code in .venv/lib/python3.12/site-packages/starlette/datastructures.py

def __init__(
    self,
    url: str = "",
    scope: Scope | None = None,
    **components: typing.Any,
) -> None:
    if scope is not None:
        assert not url, 'Cannot set both "url" and "scope".'
        assert not components, 'Cannot set both "scope" and "**components".'
        scheme = scope.get("scheme", "http")
        server = scope.get("server", None)
        path = scope["path"]
        query_string = scope.get("query_string", b"")

        host_header = None
        for key, value in scope["headers"]:
            if key == b"host":
                host_header = value.decode("latin-1")
                break

        if host_header is not None:
            url = f"{scheme}://{host_header}{path}"
        elif server is None:
            url = path
        else:
            host, port = server
            default_port = {"http": 80, "https": 443, "ws": 80, "wss": 443}[scheme]
            if port == default_port:
                url = f"{scheme}://{host}{path}"
            else:
                url = f"{scheme}://{host}:{port}{path}"

        if query_string:
            url += "?" + query_string.decode()
    elif components:
        assert not url, 'Cannot set both "url" and "**components".'
        url = URL("").replace(**components).components.geturl()

    self._url = url

Address

Bases: NamedTuple

FormData

FormData(
    *args: FormData
    | Mapping[str, str | UploadFile]
    | list[tuple[str, str | UploadFile]],
    **kwargs: str | UploadFile,
)

Bases: ImmutableMultiDict[str, Union[UploadFile, str]]

An immutable multidict, containing both file uploads and text input.

Source code in .venv/lib/python3.12/site-packages/starlette/datastructures.py

def __init__(
    self,
    *args: FormData
    | typing.Mapping[str, str | UploadFile]
    | list[tuple[str, str | UploadFile]],
    **kwargs: str | UploadFile,
) -> None:
    super().__init__(*args, **kwargs)

Headers

Headers(
    headers: Mapping[str, str] | None = None,
    raw: list[tuple[bytes, bytes]] | None = None,
    scope: MutableMapping[str, Any] | None = None,
)

Bases: Mapping[str, str]

An immutable, case-insensitive multidict.

Source code in .venv/lib/python3.12/site-packages/starlette/datastructures.py

def __init__(
    self,
    headers: typing.Mapping[str, str] | None = None,
    raw: list[tuple[bytes, bytes]] | None = None,
    scope: typing.MutableMapping[str, typing.Any] | None = None,
) -> None:
    self._list: list[tuple[bytes, bytes]] = []
    if headers is not None:
        assert raw is None, 'Cannot set both "headers" and "raw".'
        assert scope is None, 'Cannot set both "headers" and "scope".'
        self._list = [
            (key.lower().encode("latin-1"), value.encode("latin-1"))
            for key, value in headers.items()
        ]
    elif raw is not None:
        assert scope is None, 'Cannot set both "raw" and "scope".'
        self._list = raw
    elif scope is not None:
        # scope["headers"] isn't necessarily a list
        # it might be a tuple or other iterable
        self._list = scope["headers"] = list(scope["headers"])

QueryParams

QueryParams(
    *args: ImmutableMultiDict[Any, Any]
    | Mapping[Any, Any]
    | list[tuple[Any, Any]]
    | str
    | bytes,
    **kwargs: Any,
)

Bases: ImmutableMultiDict[str, str]

An immutable multidict.

Source code in .venv/lib/python3.12/site-packages/starlette/datastructures.py

def __init__(
    self,
    *args: ImmutableMultiDict[typing.Any, typing.Any]
    | typing.Mapping[typing.Any, typing.Any]
    | list[tuple[typing.Any, typing.Any]]
    | str
    | bytes,
    **kwargs: typing.Any,
) -> None:
    assert len(args) < 2, "Too many arguments."

    value = args[0] if args else []

    if isinstance(value, str):
        super().__init__(parse_qsl(value, keep_blank_values=True), **kwargs)
    elif isinstance(value, bytes):
        super().__init__(
            parse_qsl(value.decode("latin-1"), keep_blank_values=True), **kwargs
        )
    else:
        super().__init__(*args, **kwargs)  # type: ignore[arg-type]
    self._list = [(str(k), str(v)) for k, v in self._list]
    self._dict = {str(k): str(v) for k, v in self._dict.items()}

State

State(state: dict[str, Any] | None = None)

An object that can be used to store arbitrary state.

Used for request.state and app.state.

Source code in .venv/lib/python3.12/site-packages/starlette/datastructures.py

def __init__(self, state: dict[str, typing.Any] | None = None):
    if state is None:
        state = {}
    super().__setattr__("_state", state)

UploadFile

UploadFile(
    file: BinaryIO,
    *,
    size: int | None = None,
    filename: str | None = None,
    headers: Headers | None = None,
)

Bases: UploadFile

A file uploaded in a request.

Define it as a path operation function (or dependency) parameter.

If you are using a regular def function, you can use the upload_file.file attribute to access the raw standard Python file (blocking, not async), useful and needed for non-async code.

Read more about it in the FastAPI docs for Request Files.

Example

from typing import Annotated

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}

Source code in .venv/lib/python3.12/site-packages/starlette/datastructures.py

def __init__(
    self,
    file: typing.BinaryIO,
    *,
    size: int | None = None,
    filename: str | None = None,
    headers: Headers | None = None,
) -> None:
    self.filename = filename
    self.file = file
    self.size = size
    self.headers = headers or Headers()

write async

write(
    data: Annotated[
        bytes,
        Doc(
            "\n                The bytes to write to the file.\n                "
        ),
    ],
) -> None

Write some bytes to the file.

You normally wouldn't use this from a file you read in a request.

To be awaitable, compatible with async, this is run in threadpool.

Source code in .venv/lib/python3.12/site-packages/fastapi/datastructures.py

async def write(
    self,
    data: Annotated[
        bytes,
        Doc(
            """
            The bytes to write to the file.
            """
        ),
    ],
) -> None:
    """
    Write some bytes to the file.

    You normally wouldn't use this from a file you read in a request.

    To be awaitable, compatible with async, this is run in threadpool.
    """
    return await super().write(data)

read async

read(
    size: Annotated[
        int,
        Doc(
            "\n                The number of bytes to read from the file.\n                "
        ),
    ] = -1,
) -> bytes

Read some bytes from the file.

To be awaitable, compatible with async, this is run in threadpool.

Source code in .venv/lib/python3.12/site-packages/fastapi/datastructures.py

async def read(
    self,
    size: Annotated[
        int,
        Doc(
            """
            The number of bytes to read from the file.
            """
        ),
    ] = -1,
) -> bytes:
    """
    Read some bytes from the file.

    To be awaitable, compatible with async, this is run in threadpool.
    """
    return await super().read(size)

seek async

seek(
    offset: Annotated[
        int,
        Doc(
            "\n                The position in bytes to seek to in the file.\n                "
        ),
    ],
) -> None

Move to a position in the file.

Any next read or write will be done from that position.

To be awaitable, compatible with async, this is run in threadpool.

Source code in .venv/lib/python3.12/site-packages/fastapi/datastructures.py

async def seek(
    self,
    offset: Annotated[
        int,
        Doc(
            """
            The position in bytes to seek to in the file.
            """
        ),
    ],
) -> None:
    """
    Move to a position in the file.

    Any next read or write will be done from that position.

    To be awaitable, compatible with async, this is run in threadpool.
    """
    return await super().seek(offset)

close async

close() -> None

Close the file.

To be awaitable, compatible with async, this is run in threadpool.

Source code in .venv/lib/python3.12/site-packages/fastapi/datastructures.py

async def close(self) -> None:
    """
    Close the file.

    To be awaitable, compatible with async, this is run in threadpool.
    """
    return await super().close()

plateforme.core.api.requests

This module provides utilities for managing requests within the Plateforme framework's API using FastAPI and Starlette features.

HTTPConnection

HTTPConnection(
    scope: Scope, receive: Receive | None = None
)

Bases: Mapping[str, Any]

A base class for incoming HTTP connections, that is used to provide any functionality that is common to both Request and WebSocket.

Source code in .venv/lib/python3.12/site-packages/starlette/requests.py

def __init__(self, scope: Scope, receive: Receive | None = None) -> None:
    assert scope["type"] in ("http", "websocket")
    self.scope = scope

Request

Request(
    scope: Scope,
    receive: Receive = empty_receive,
    send: Send = empty_send,
)

Bases: HTTPConnection

Source code in .venv/lib/python3.12/site-packages/starlette/requests.py

def __init__(
    self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send
):
    super().__init__(scope)
    assert scope["type"] == "http"
    self._receive = receive
    self._send = send
    self._stream_consumed = False
    self._is_disconnected = False
    self._form = None

plateforme.core.api.responses

This module provides utilities for managing responses within the Plateforme framework's API using FastAPI and Starlette features.

Response

Response(
    content: Any = None,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
)

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    content: typing.Any = None,
    status_code: int = 200,
    headers: typing.Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
) -> None:
    self.status_code = status_code
    if media_type is not None:
        self.media_type = media_type
    self.background = background
    self.body = self.render(content)
    self.init_headers(headers)

FileResponse

FileResponse(
    path: str | PathLike[str],
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
    filename: str | None = None,
    stat_result: stat_result | None = None,
    method: str | None = None,
    content_disposition_type: str = "attachment",
)

Bases: Response

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    path: str | os.PathLike[str],
    status_code: int = 200,
    headers: typing.Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
    filename: str | None = None,
    stat_result: os.stat_result | None = None,
    method: str | None = None,
    content_disposition_type: str = "attachment",
) -> None:
    self.path = path
    self.status_code = status_code
    self.filename = filename
    if method is not None:
        warnings.warn(
            "The 'method' parameter is not used, and it will be removed.",
            DeprecationWarning,
        )
    if media_type is None:
        media_type = guess_type(filename or path)[0] or "text/plain"
    self.media_type = media_type
    self.background = background
    self.init_headers(headers)
    if self.filename is not None:
        content_disposition_filename = quote(self.filename)
        if content_disposition_filename != self.filename:
            content_disposition = "{}; filename*=utf-8''{}".format(
                content_disposition_type, content_disposition_filename
            )
        else:
            content_disposition = '{}; filename="{}"'.format(
                content_disposition_type, self.filename
            )
        self.headers.setdefault("content-disposition", content_disposition)
    self.stat_result = stat_result
    if stat_result is not None:
        self.set_stat_headers(stat_result)

HTMLResponse

HTMLResponse(
    content: Any = None,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
)

Bases: Response

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    content: typing.Any = None,
    status_code: int = 200,
    headers: typing.Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
) -> None:
    self.status_code = status_code
    if media_type is not None:
        self.media_type = media_type
    self.background = background
    self.body = self.render(content)
    self.init_headers(headers)

JSONResponse

JSONResponse(
    content: Any,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
)

Bases: Response

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    content: typing.Any,
    status_code: int = 200,
    headers: typing.Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
) -> None:
    super().__init__(content, status_code, headers, media_type, background)

ORJSONResponse

ORJSONResponse(
    content: Any,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
)

Bases: JSONResponse

JSON response using the high-performance orjson library to serialize data to JSON.

Read more about it in the FastAPI docs for Custom Response - HTML, Stream, File, others.

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    content: typing.Any,
    status_code: int = 200,
    headers: typing.Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
) -> None:
    super().__init__(content, status_code, headers, media_type, background)

PlainTextResponse

PlainTextResponse(
    content: Any = None,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
)

Bases: Response

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    content: typing.Any = None,
    status_code: int = 200,
    headers: typing.Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
) -> None:
    self.status_code = status_code
    if media_type is not None:
        self.media_type = media_type
    self.background = background
    self.body = self.render(content)
    self.init_headers(headers)

RedirectResponse

RedirectResponse(
    url: str | URL,
    status_code: int = 307,
    headers: Mapping[str, str] | None = None,
    background: BackgroundTask | None = None,
)

Bases: Response

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    url: str | URL,
    status_code: int = 307,
    headers: typing.Mapping[str, str] | None = None,
    background: BackgroundTask | None = None,
) -> None:
    super().__init__(
        content=b"", status_code=status_code, headers=headers, background=background
    )
    self.headers["location"] = quote(str(url), safe=":/%#?=@[]!$&'()*+,;")

StreamingResponse

StreamingResponse(
    content: ContentStream,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
)

Bases: Response

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    content: ContentStream,
    status_code: int = 200,
    headers: typing.Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
) -> None:
    if isinstance(content, typing.AsyncIterable):
        self.body_iterator = content
    else:
        self.body_iterator = iterate_in_threadpool(content)
    self.status_code = status_code
    self.media_type = self.media_type if media_type is None else media_type
    self.background = background
    self.init_headers(headers)

UJSONResponse

UJSONResponse(
    content: Any,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
)

Bases: JSONResponse

JSON response using the high-performance ujson library to serialize data to JSON.

Read more about it in the FastAPI docs for Custom Response - HTML, Stream, File, others.

Source code in .venv/lib/python3.12/site-packages/starlette/responses.py

def __init__(
    self,
    content: typing.Any,
    status_code: int = 200,
    headers: typing.Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
) -> None:
    super().__init__(content, status_code, headers, media_type, background)

plateforme.core.api.websockets

This module provides utilities for managing websockets within the Plateforme framework's API using FastAPI and Starlette features.

WebSocket

WebSocket(scope: Scope, receive: Receive, send: Send)

Bases: HTTPConnection

Source code in .venv/lib/python3.12/site-packages/starlette/websockets.py

def __init__(self, scope: Scope, receive: Receive, send: Send) -> None:
    super().__init__(scope)
    assert scope["type"] == "websocket"
    self._receive = receive
    self._send = send
    self.client_state = WebSocketState.CONNECTING
    self.application_state = WebSocketState.CONNECTING

receive async

receive() -> Message

Receive ASGI websocket messages, ensuring valid state transitions.

Source code in .venv/lib/python3.12/site-packages/starlette/websockets.py

async def receive(self) -> Message:
    """
    Receive ASGI websocket messages, ensuring valid state transitions.
    """
    if self.client_state == WebSocketState.CONNECTING:
        message = await self._receive()
        message_type = message["type"]
        if message_type != "websocket.connect":
            raise RuntimeError(
                'Expected ASGI message "websocket.connect", '
                f"but got {message_type!r}"
            )
        self.client_state = WebSocketState.CONNECTED
        return message
    elif self.client_state == WebSocketState.CONNECTED:
        message = await self._receive()
        message_type = message["type"]
        if message_type not in {"websocket.receive", "websocket.disconnect"}:
            raise RuntimeError(
                'Expected ASGI message "websocket.receive" or '
                f'"websocket.disconnect", but got {message_type!r}'
            )
        if message_type == "websocket.disconnect":
            self.client_state = WebSocketState.DISCONNECTED
        return message
    else:
        raise RuntimeError(
            'Cannot call "receive" once a disconnect message has been received.'
        )

send async

send(message: Message) -> None

Send ASGI websocket messages, ensuring valid state transitions.

Source code in .venv/lib/python3.12/site-packages/starlette/websockets.py

async def send(self, message: Message) -> None:
    """
    Send ASGI websocket messages, ensuring valid state transitions.
    """
    if self.application_state == WebSocketState.CONNECTING:
        message_type = message["type"]
        if message_type not in {"websocket.accept", "websocket.close"}:
            raise RuntimeError(
                'Expected ASGI message "websocket.accept" or '
                f'"websocket.close", but got {message_type!r}'
            )
        if message_type == "websocket.close":
            self.application_state = WebSocketState.DISCONNECTED
        else:
            self.application_state = WebSocketState.CONNECTED
        await self._send(message)
    elif self.application_state == WebSocketState.CONNECTED:
        message_type = message["type"]
        if message_type not in {"websocket.send", "websocket.close"}:
            raise RuntimeError(
                'Expected ASGI message "websocket.send" or "websocket.close", '
                f"but got {message_type!r}"
            )
        if message_type == "websocket.close":
            self.application_state = WebSocketState.DISCONNECTED
        try:
            await self._send(message)
        except IOError:
            self.application_state = WebSocketState.DISCONNECTED
            raise WebSocketDisconnect(code=1006)
    else:
        raise RuntimeError('Cannot call "send" once a close message has been sent.')

WebSocketDisconnect

WebSocketDisconnect(
    code: int = 1000, reason: Optional[str] = None
)

Bases: Exception

Source code in .venv/lib/python3.12/site-packages/starlette/websockets.py

def __init__(self, code: int = 1000, reason: typing.Optional[str] = None) -> None:
    self.code = code
    self.reason = reason or ""