MCPcopy Create free account
hub / github.com/encode/starlette / FileResponse

Class FileResponse

starlette/responses.py:298–579  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

296
297
298class FileResponse(Response):
299 chunk_size = 64 * 1024
300 max_ranges = 100
301
302 def __init__(
303 self,
304 path: str | os.PathLike[str],
305 status_code: int = 200,
306 headers: Mapping[str, str] | None = None,
307 media_type: str | None = None,
308 background: BackgroundTask | None = None,
309 filename: str | None = None,
310 stat_result: os.stat_result | None = None,
311 content_disposition_type: str = "attachment",
312 ) -> None:
313 self.path = path
314 self.status_code = status_code
315 self.filename = filename
316 if media_type is None:
317 media_type = guess_type(filename or path)[0] or "application/octet-stream"
318 self.media_type = media_type
319 self.background = background
320 self.init_headers(headers)
321 self.headers.setdefault("accept-ranges", "bytes")
322 if self.filename is not None:
323 content_disposition_filename = quote(self.filename)
324 if content_disposition_filename != self.filename:
325 content_disposition = f"{content_disposition_type}; filename*=utf-8''{content_disposition_filename}"
326 else:
327 content_disposition = f'{content_disposition_type}; filename="{self.filename}"'
328 self.headers.setdefault("content-disposition", content_disposition)
329 self.stat_result = stat_result
330 if stat_result is not None:
331 self.set_stat_headers(stat_result)
332
333 def set_stat_headers(self, stat_result: os.stat_result) -> None:
334 content_length = str(stat_result.st_size)
335 last_modified = formatdate(stat_result.st_mtime, usegmt=True)
336 etag_base = str(stat_result.st_mtime) + "-" + str(stat_result.st_size)
337 etag = f'"{hashlib.md5(etag_base.encode(), usedforsecurity=False).hexdigest()}"'
338
339 self.headers.setdefault("content-length", content_length)
340 self.headers.setdefault("last-modified", last_modified)
341 self.headers.setdefault("etag", etag)
342
343 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
344 scope_type = scope["type"]
345 send_header_only = scope_type == "http" and scope["method"].upper() == "HEAD"
346 send_pathsend = scope_type == "http" and "http.response.pathsend" in scope.get("extensions", {})
347 if scope_type == "websocket":
348 send = self._wrap_websocket_denial_send(send)
349
350 if self.stat_result is None:
351 try:
352 stat_result = await anyio.to_thread.run_sync(os.stat, self.path)
353 self.set_stat_headers(stat_result)
354 except FileNotFoundError:
355 raise RuntimeError(f"File at path {self.path} does not exist.")

Calls

no outgoing calls