# Files and Streams

> UCall for Python

Handlers run on the polling thread, so a reply or a pull that finishes later runs on a worker: `request.detach()` hands the request over as a `ucall.DetachedRequest`, which reads and replies like a `Request` and alone has the three verbs below.

| Verb                                  | Returns                                             | Final when                             |
| :------------------------------------ | :-------------------------------------------------- | :------------------------------------- |
| `reply_parts(parts, status, headers)` | `None` for bytes alone, else a `ucall.Deferred`     | every part is sent                     |
| `reply_stream(status, headers)`       | a `ucall.ReplyStream` with `append(parts)`, `end()` | `end()` returns `None` or a `Deferred` |
| `body_chunks()`                       | an iterator and async iterator of `memoryview`s     | the body ended                         |

### Serving a File

```python
import concurrent.futures

import ucall

workers = concurrent.futures.ThreadPoolExecutor()
server = ucall.Server(ucall.Interface("0.0.0.0:8080", reply_directories=["/srv/videos"]))

def send_video(detached: ucall.DetachedRequest, name: str):
    deferred = detached.reply_parts([ucall.Path(0, name)])
    deferred.wait()  # or: await deferred

@server.get("/videos/{name}")
def video(request: ucall.Request):
    workers.submit(send_video, request.detach(), request.path_parameter("name"))
```

`reply_parts` concatenates bytes-like objects, `ucall.Path(directory, path, offset=0, length=0)` and `ucall.File(descriptor, offset=0, length=0)` into one body whose Content-Length is measured from the files.
A `Path` resolves beneath the named `reply_directories` entry, and the event loop opens, reads and closes it unbuffered.
A `File` is the caller's regular file opened with `os.O_DIRECT`, or an object whose `fileno()` answers one; UCall never closes it, and the `Deferred` keeps it open until the reply is final.
`deferred.status()` reads the outcome without waiting, `deferred.wait()` blocks without the interpreter lock and raises the failure, such as `Status.NOT_FOUND` after a 404, and `await deferred` runs that wait on the running loop's default executor.
Releasing an unfinished `Deferred` waits for it, which is why only a detached request makes one.

### Streaming a Reply

```python
def send_rows(detached: ucall.DetachedRequest, rows):
    stream = detached.reply_stream(headers=("content-type: text/csv",))
    for row in rows:
        stream.append([row.encode()])
    stream.end()
```

`append` queues more parts and raises `ucall.Error` with `Status.WOULD_BLOCK` while the peer lags behind what is queued; `end()` answers `None` or a `Deferred` as `reply_parts` does, and a stream released without `end()` ends itself.

### Streaming an Upload

```python
import hashlib

def store(detached: ucall.DetachedRequest):
    digest = hashlib.sha256()
    for chunk in detached.body_chunks():  # or: async for chunk in detached.body_chunks()
        digest.update(chunk)
    detached.reply(digest.hexdigest().encode())

@server.post("/uploads", body=ucall.BodyDelivery.STREAMED)
def upload(request: ucall.Request):
    workers.submit(store, request.detach())
```

An endpoint registered with `body=ucall.BodyDelivery.STREAMED`, on `add_endpoint`, `route` and every verb decorator, runs its handler once the headers arrived.
`body_chunks()` yields each chunk as a `memoryview` valid until the next pull, which releases it, and waits for bytes without the interpreter lock.
`async for` over it returns a chunk already decoded at once and runs the wait for one still arriving on the running loop's default executor, as `await deferred` does.

`from ucall import File, Path, ReplyStream`

## File

A byte range of a file the caller opened, as one part of [`DetachedRequest.reply_parts`](https://www.unum.cloud/docs/ucall/python/requests.md#reply_parts).

```python
class File
```

The event loop reads it unbuffered after the call returns, so the reply defers, and the
[`Deferred`](https://www.unum.cloud/docs/ucall/python/requests.md#Deferred) keeps this part, and through it a file object, alive until it is final. UCall
never closes the descriptor.

Examples: ```
>>> part = ucall.File(os.open("/srv/videos/intro.mp4", os.O_RDONLY | os.O_DIRECT))
```

## File

```python
File(descriptor: Any, offset: int = 0, length: int = 0)
```

## descriptor

The descriptor or its file object.

```python
descriptor: Any
```

## offset

The range's first byte.

```python
offset: int
```

## length

The range's bytes; 0 runs to the end.

```python
length: int
```

## Path

A byte range of a file beneath a reply directory, as one part of [`reply_parts`](https://www.unum.cloud/docs/ucall/python/requests.md#reply_parts).

```python
class Path
```

The event loop opens it unbuffered beneath the directory and closes it once sent, so the
reply defers. Absolute paths, ".." and any resolution leaving the directory are refused.

Examples: ```
>>> part = ucall.Path(0, "intro.mp4", offset=1 << 20)
```

## Path

```python
Path(
    directory: int,
    path: str | bytes | os.PathLike[str] | os.PathLike[bytes],
    offset: int = 0,
    length: int = 0,
)
```

## directory

The position of the reply directory.

```python
directory: int
```

## path

The file beneath the directory.

```python
path: str | bytes
```

## offset

The range's first byte.

```python
offset: int
```

## length

The range's bytes; 0 runs to the end.

```python
length: int
```

## ReplyStream

A reply sent in parts, from [`DetachedRequest.reply_stream`](https://www.unum.cloud/docs/ucall/python/requests.md#reply_stream), ended once.

```python
class ReplyStream
```

Released without `end()`, it ends the reply itself and waits for any file parts to send.

Examples: ```
>>> stream = detached.reply_stream(headers=("content-type: application/x-ndjson",))
>>> for row in rows:
...     stream.append([json.dumps(row).encode(), b"\n"])
>>> stream.end()
```

## append

Queues [`parts`](https://www.unum.cloud/docs/ucall/python/requests.md#parts) after the ones before, all of them or none.

```python
def append(parts: Sequence[Buffer | File | Path], /) -> None
```

Bytes-like parts are copied before the call returns; [`File`](#File) parts stay alive with the
stream until its end is final.

- `parts`: The next parts of the body.

Raises: - [`Error`](https://www.unum.cloud/docs/ucall/python/errors.md#Error): [`Status.WOULD_BLOCK`](https://www.unum.cloud/docs/ucall/python/errors.md#WOULD_BLOCK) while the peer lags, to retry once it caught up, or
[`Status.CONNECTION_CLOSED`](https://www.unum.cloud/docs/ucall/python/errors.md#CONNECTION_CLOSED) once it left; nothing was queued either way.

Examples: ```
>>> stream.append([b"data: ", payload, b"\n\n"])
```

## end

Completes the reply, consuming the request whatever happens.

```python
def end() -> Deferred | None
```

Raises: - [`Error`](https://www.unum.cloud/docs/ucall/python/errors.md#Error): The parts fell short of a declared Content-Length, with [`Status.MALFORMED`](https://www.unum.cloud/docs/ucall/python/errors.md#MALFORMED), or
the stream already ended, with [`Status.ALREADY_EXISTS`](https://www.unum.cloud/docs/ucall/python/errors.md#ALREADY_EXISTS).

Examples: ```
>>> deferred = stream.end()
```