We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
StaticFilesConfig and virtual directories
I'm trying to write a ``FileSystemProtocol`` to load files from the package data using [importlib_resources](https://importlib-resources.readthedocs.io/en/latest/using.html#). But because ``directories`` is defined as ``DirectoryPath``, pydantic checks if the given directories exist in the local filesystem. 

This is not generally true, especially in any kind of virtual filesystem (e.g. a zipped package). I think this condition should be relaxed to support virtual filesystems.

https://github.com/starlite-api/starlite/blob/9bb6dcd57c10a591377cf8e3a537e9292566d5b9/starlite/config/static_files.py#L32

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `tools/build_docs.py`
Content:
```
1 from __future__ import annotations
2 
3 import argparse
4 import importlib.metadata
5 import json
6 import os
7 import shutil
8 import subprocess
9 from contextlib import contextmanager
10 from pathlib import Path
11 from typing import TypedDict
12 
13 REDIRECT_TEMPLATE = """
14 <!DOCTYPE HTML>
15 <html lang="en-US">
16     <head>
17         <title>Page Redirection</title>
18         <meta charset="UTF-8">
19         <meta http-equiv="refresh" content="0; url={target}">
20         <script type="text/javascript">window.location.href = "{target}"</script>
21     </head>
22     <body>
23         You are being redirected. If this does not work, click <a href='{target}'>this link</a>
24     </body>
25 </html>
26 """
27 
28 parser = argparse.ArgumentParser()
29 parser.add_argument("--version", required=False)
30 parser.add_argument("output")
31 
32 
33 class VersionSpec(TypedDict):
34     versions: list[str]
35     latest: str
36 
37 
38 @contextmanager
39 def checkout(branch: str) -> None:
40     subprocess.run(["git", "checkout", branch], check=True)  # noqa: S603 S607
41     yield
42     subprocess.run(["git", "checkout", "-"], check=True)  # noqa: S603 S607
43 
44 
45 def load_version_spec() -> VersionSpec:
46     versions_file = Path("docs/_static/versions.json")
47     if versions_file.exists():
48         return json.loads(versions_file.read_text())
49     return {"versions": [], "latest": ""}
50 
51 
52 def build(output_dir: str, version: str | None) -> None:
53     if version is None:
54         version = importlib.metadata.version("litestar").rsplit(".")[0]
55     else:
56         os.environ["_LITESTAR_DOCS_BUILD_VERSION"] = version
57 
58     subprocess.run(["make", "docs"], check=True)  # noqa: S603 S607
59 
60     output_dir = Path(output_dir)
61     output_dir.mkdir()
62     output_dir.joinpath(".nojekyll").touch(exist_ok=True)
63 
64     version_spec = load_version_spec()
65     is_latest = version == version_spec["latest"]
66 
67     docs_src_path = Path("docs/_build/html")
68 
69     output_dir.joinpath("index.html").write_text(REDIRECT_TEMPLATE.format(target="latest"))
70 
71     if is_latest:
72         shutil.copytree(docs_src_path, output_dir / "latest", dirs_exist_ok=True)
73     shutil.copytree(docs_src_path, output_dir / version, dirs_exist_ok=True)
74 
75     # copy existing versions into our output dir to preserve them when cleaning the branch
76     with checkout("gh-pages"):
77         for other_version in [*version_spec["versions"], "latest"]:
78             other_version_path = Path(other_version)
79             other_version_target_path = output_dir / other_version
80             if other_version_path.exists() and not other_version_target_path.exists():
81                 shutil.copytree(other_version_path, other_version_target_path)
82 
83 
84 def main() -> None:
85     args = parser.parse_args()
86     build(output_dir=args.output, version=args.version)
87 
88 
89 if __name__ == "__main__":
90     main()
91
```

Path: `tools/sphinx_ext/run_examples.py`
Content:
```
1 from __future__ import annotations
2 
3 import importlib
4 import logging
5 import multiprocessing
6 import os
7 import re
8 import shlex
9 import socket
10 import subprocess
11 import sys
12 import time
13 from contextlib import contextmanager, redirect_stderr
14 from pathlib import Path
15 from typing import TYPE_CHECKING, Generator
16 
17 import httpx
18 import uvicorn
19 from auto_pytabs.sphinx_ext import CodeBlockOverride, LiteralIncludeOverride
20 from docutils.nodes import Node, admonition, literal_block, title
21 from docutils.parsers.rst import directives
22 from sphinx.addnodes import highlightlang
23 
24 from litestar import Litestar
25 
26 if TYPE_CHECKING:
27     from sphinx.application import Sphinx
28 
29 
30 RGX_RUN = re.compile(r"# +?run:(.*)")
31 
32 AVAILABLE_PORTS = list(range(9000, 9999))
33 
34 
35 logger = logging.getLogger("sphinx")
36 
37 
38 def _load_app_from_path(path: Path) -> Litestar:
39     module = importlib.import_module(str(path.with_suffix("")).replace("/", "."))
40     for obj in module.__dict__.values():
41         if isinstance(obj, Litestar):
42             return obj
43     raise RuntimeError(f"No Litestar app found in {path}")
44 
45 
46 @contextmanager
47 def run_app(path: Path) -> Generator[int, None, None]:
48     """Run an example app from a python file.
49 
50     The first ``Litestar`` instance found in the file will be used as target to run.
51     """
52     while AVAILABLE_PORTS:
53         port = AVAILABLE_PORTS.pop()
54         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
55             if sock.connect_ex(("127.0.0.1", port)) != 0:
56                 break
57     else:
58         raise RuntimeError("Could not find an open port")
59 
60     app = _load_app_from_path(path)
61 
62     def run() -> None:
63         with redirect_stderr(Path(os.devnull).open()):
64             uvicorn.run(app, port=port, access_log=False)
65 
66     proc = multiprocessing.Process(target=run)
67     proc.start()
68     for _ in range(50):
69         try:
70             httpx.get(f"http://127.0.0.1:{port}", timeout=0.1)
71         except httpx.TransportError:
72             time.sleep(0.1)
73         else:
74             break
75     try:
76         yield port
77     finally:
78         proc.kill()
79     AVAILABLE_PORTS.append(port)
80 
81 
82 def extract_run_args(content: str) -> tuple[str, list[list[str]]]:
83     """Extract run args from a python file.
84 
85     Return the file content stripped of the run comments and a list of argument lists
86     """
87     new_lines = []
88     run_configs = []
89     for line in content.splitlines():
90         if run_stmt_match := RGX_RUN.match(line):
91             run_stmt = run_stmt_match.group(1).lstrip()
92             run_configs.append(shlex.split(run_stmt))
93         else:
94             new_lines.append(line)
95     return "\n".join(new_lines), run_configs
96 
97 
98 def exec_examples(app_file: Path, run_configs: list[list[str]]) -> str:
99     """Start a server with the example application, run the specified requests against it
100     and return their results
101     """
102 
103     results = []
104 
105     with run_app(app_file) as port:
106         for run_args in run_configs:
107             url_path, *options = run_args
108             args = ["curl", "-s", f"http://127.0.0.1:{port}{url_path}", *options]
109             clean_args = ["curl", f"http://127.0.0.1:8000{url_path}", *options]
110 
111             proc = subprocess.run(
112                 args,  # noqa: S603
113                 capture_output=True,
114                 text=True,
115             )
116             stdout = proc.stdout.splitlines()
117             if not stdout:
118                 logger.error(f"Example: {app_file}:{args} yielded no results")
119                 continue
120 
121             result = "\n".join(line for line in ("> " + (" ".join(clean_args)), *stdout))
122             results.append(result)
123 
124     return "\n".join(results)
125 
126 
127 class LiteralInclude(LiteralIncludeOverride):
128     option_spec = {**LiteralIncludeOverride.option_spec, "no-run": directives.flag}
129 
130     def run(self) -> list[Node]:
131         cwd = Path.cwd()
132         docs_dir = cwd / "docs"
133         language = self.options.get("language")
134         file = Path(self.env.relfn2path(self.arguments[0])[1])
135 
136         if (language != "python" and file.suffix != ".py") or "no-run" in self.options:
137             return super().run()
138 
139         content = file.read_text()
140         clean_content, run_args = extract_run_args(content)
141 
142         if not run_args:
143             return super().run()
144 
145         tmp_file = self.env.tmp_examples_path / str(file.relative_to(docs_dir)).replace("/", "_")
146 
147         self.arguments[0] = "/" + str(tmp_file.relative_to(docs_dir))
148         tmp_file.write_text(clean_content)
149 
150         nodes = super().run()
151 
152         result = exec_examples(file.relative_to(cwd), run_args)
153 
154         nodes.append(
155             admonition(
156                 "",
157                 title("", "Run it"),
158                 highlightlang(
159                     "",
160                     literal_block("", result),
161                     lang="shell",
162                     force=False,
163                     linenothreshold=sys.maxsize,
164                 ),
165                 literal_block("", result),
166             )
167         )
168 
169         return nodes
170 
171 
172 def setup(app: Sphinx) -> dict[str, bool]:
173     app.add_directive("literalinclude", LiteralInclude, override=True)
174     app.add_directive("code-block", CodeBlockOverride, override=True)
175 
176     return {"parallel_read_safe": True, "parallel_write_safe": True}
177
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/tools/build_docs.py b/tools/build_docs.py
--- a/tools/build_docs.py
+++ b/tools/build_docs.py
@@ -27,6 +27,7 @@
 
 parser = argparse.ArgumentParser()
 parser.add_argument("--version", required=False)
+parser.add_argument("--ignore-missing-examples-output", action="store_true", default=False)
 parser.add_argument("output")
 
 
@@ -49,12 +50,15 @@
     return {"versions": [], "latest": ""}
 
 
-def build(output_dir: str, version: str | None) -> None:
+def build(output_dir: str, version: str | None, ignore_missing_output: bool) -> None:
     if version is None:
         version = importlib.metadata.version("litestar").rsplit(".")[0]
     else:
         os.environ["_LITESTAR_DOCS_BUILD_VERSION"] = version
 
+    if ignore_missing_output:
+        os.environ["_LITESTAR_DOCS_IGNORE_MISSING_EXAMPLE_OUTPUT"] = "1"
+
     subprocess.run(["make", "docs"], check=True)  # noqa: S603 S607
 
     output_dir = Path(output_dir)
@@ -83,7 +87,11 @@
 
 def main() -> None:
     args = parser.parse_args()
-    build(output_dir=args.output, version=args.version)
+    build(
+        output_dir=args.output,
+        version=args.version,
+        ignore_missing_output=args.ignore_missing_output,
+    )
 
 
 if __name__ == "__main__":
diff --git a/tools/sphinx_ext/run_examples.py b/tools/sphinx_ext/run_examples.py
--- a/tools/sphinx_ext/run_examples.py
+++ b/tools/sphinx_ext/run_examples.py
@@ -34,6 +34,8 @@
 
 logger = logging.getLogger("sphinx")
 
+ignore_missing_output = os.getenv("_LITESTAR_DOCS_IGNORE_MISSING_EXAMPLE_OUTPUT", "") == "1"
+
 
 def _load_app_from_path(path: Path) -> Litestar:
     module = importlib.import_module(str(path.with_suffix("")).replace("/", "."))
@@ -115,7 +117,9 @@
             )
             stdout = proc.stdout.splitlines()
             if not stdout:
-                logger.error(f"Example: {app_file}:{args} yielded no results")
+                logger.debug(proc.stderr)
+                if not ignore_missing_output:
+                    logger.error(f"Example: {app_file}:{args} yielded no results")
                 continue
 
             result = "\n".join(line for line in ("> " + (" ".join(clean_args)), *stdout))

{"golden_diff": "diff --git a/tools/build_docs.py b/tools/build_docs.py\n--- a/tools/build_docs.py\n+++ b/tools/build_docs.py\n@@ -27,6 +27,7 @@\n \n parser = argparse.ArgumentParser()\n parser.add_argument(\"--version\", required=False)\n+parser.add_argument(\"--ignore-missing-examples-output\", action=\"store_true\", default=False)\n parser.add_argument(\"output\")\n \n \n@@ -49,12 +50,15 @@\n     return {\"versions\": [], \"latest\": \"\"}\n \n \n-def build(output_dir: str, version: str | None) -> None:\n+def build(output_dir: str, version: str | None, ignore_missing_output: bool) -> None:\n     if version is None:\n         version = importlib.metadata.version(\"litestar\").rsplit(\".\")[0]\n     else:\n         os.environ[\"_LITESTAR_DOCS_BUILD_VERSION\"] = version\n \n+    if ignore_missing_output:\n+        os.environ[\"_LITESTAR_DOCS_IGNORE_MISSING_EXAMPLE_OUTPUT\"] = \"1\"\n+\n     subprocess.run([\"make\", \"docs\"], check=True)  # noqa: S603 S607\n \n     output_dir = Path(output_dir)\n@@ -83,7 +87,11 @@\n \n def main() -> None:\n     args = parser.parse_args()\n-    build(output_dir=args.output, version=args.version)\n+    build(\n+        output_dir=args.output,\n+        version=args.version,\n+        ignore_missing_output=args.ignore_missing_output,\n+    )\n \n \n if __name__ == \"__main__\":\ndiff --git a/tools/sphinx_ext/run_examples.py b/tools/sphinx_ext/run_examples.py\n--- a/tools/sphinx_ext/run_examples.py\n+++ b/tools/sphinx_ext/run_examples.py\n@@ -34,6 +34,8 @@\n \n logger = logging.getLogger(\"sphinx\")\n \n+ignore_missing_output = os.getenv(\"_LITESTAR_DOCS_IGNORE_MISSING_EXAMPLE_OUTPUT\", \"\") == \"1\"\n+\n \n def _load_app_from_path(path: Path) -> Litestar:\n     module = importlib.import_module(str(path.with_suffix(\"\")).replace(\"/\", \".\"))\n@@ -115,7 +117,9 @@\n             )\n             stdout = proc.stdout.splitlines()\n             if not stdout:\n-                logger.error(f\"Example: {app_file}:{args} yielded no results\")\n+                logger.debug(proc.stderr)\n+                if not ignore_missing_output:\n+                    logger.error(f\"Example: {app_file}:{args} yielded no results\")\n                 continue\n \n             result = \"\\n\".join(line for line in (\"> \" + (\" \".join(clean_args)), *stdout))\n", "issue": "StaticFilesConfig and virtual directories\nI'm trying to write a ``FileSystemProtocol`` to load files from the package data using [importlib_resources](https://importlib-resources.readthedocs.io/en/latest/using.html#). But because ``directories`` is defined as ``DirectoryPath``, pydantic checks if the given directories exist in the local filesystem. \r\n\r\nThis is not generally true, especially in any kind of virtual filesystem (e.g. a zipped package). I think this condition should be relaxed to support virtual filesystems.\r\n\r\nhttps://github.com/starlite-api/starlite/blob/9bb6dcd57c10a591377cf8e3a537e9292566d5b9/starlite/config/static_files.py#L32\n", "before_files": [{"content": "from __future__ import annotations\n\nimport argparse\nimport importlib.metadata\nimport json\nimport os\nimport shutil\nimport subprocess\nfrom contextlib import contextmanager\nfrom pathlib import Path\nfrom typing import TypedDict\n\nREDIRECT_TEMPLATE = \"\"\"\n<!DOCTYPE HTML>\n<html lang=\"en-US\">\n    <head>\n        <title>Page Redirection</title>\n        <meta charset=\"UTF-8\">\n        <meta http-equiv=\"refresh\" content=\"0; url={target}\">\n        <script type=\"text/javascript\">window.location.href = \"{target}\"</script>\n    </head>\n    <body>\n        You are being redirected. If this does not work, click <a href='{target}'>this link</a>\n    </body>\n</html>\n\"\"\"\n\nparser = argparse.ArgumentParser()\nparser.add_argument(\"--version\", required=False)\nparser.add_argument(\"output\")\n\n\nclass VersionSpec(TypedDict):\n    versions: list[str]\n    latest: str\n\n\n@contextmanager\ndef checkout(branch: str) -> None:\n    subprocess.run([\"git\", \"checkout\", branch], check=True)  # noqa: S603 S607\n    yield\n    subprocess.run([\"git\", \"checkout\", \"-\"], check=True)  # noqa: S603 S607\n\n\ndef load_version_spec() -> VersionSpec:\n    versions_file = Path(\"docs/_static/versions.json\")\n    if versions_file.exists():\n        return json.loads(versions_file.read_text())\n    return {\"versions\": [], \"latest\": \"\"}\n\n\ndef build(output_dir: str, version: str | None) -> None:\n    if version is None:\n        version = importlib.metadata.version(\"litestar\").rsplit(\".\")[0]\n    else:\n        os.environ[\"_LITESTAR_DOCS_BUILD_VERSION\"] = version\n\n    subprocess.run([\"make\", \"docs\"], check=True)  # noqa: S603 S607\n\n    output_dir = Path(output_dir)\n    output_dir.mkdir()\n    output_dir.joinpath(\".nojekyll\").touch(exist_ok=True)\n\n    version_spec = load_version_spec()\n    is_latest = version == version_spec[\"latest\"]\n\n    docs_src_path = Path(\"docs/_build/html\")\n\n    output_dir.joinpath(\"index.html\").write_text(REDIRECT_TEMPLATE.format(target=\"latest\"))\n\n    if is_latest:\n        shutil.copytree(docs_src_path, output_dir / \"latest\", dirs_exist_ok=True)\n    shutil.copytree(docs_src_path, output_dir / version, dirs_exist_ok=True)\n\n    # copy existing versions into our output dir to preserve them when cleaning the branch\n    with checkout(\"gh-pages\"):\n        for other_version in [*version_spec[\"versions\"], \"latest\"]:\n            other_version_path = Path(other_version)\n            other_version_target_path = output_dir / other_version\n            if other_version_path.exists() and not other_version_target_path.exists():\n                shutil.copytree(other_version_path, other_version_target_path)\n\n\ndef main() -> None:\n    args = parser.parse_args()\n    build(output_dir=args.output, version=args.version)\n\n\nif __name__ == \"__main__\":\n    main()\n", "path": "tools/build_docs.py"}, {"content": "from __future__ import annotations\n\nimport importlib\nimport logging\nimport multiprocessing\nimport os\nimport re\nimport shlex\nimport socket\nimport subprocess\nimport sys\nimport time\nfrom contextlib import contextmanager, redirect_stderr\nfrom pathlib import Path\nfrom typing import TYPE_CHECKING, Generator\n\nimport httpx\nimport uvicorn\nfrom auto_pytabs.sphinx_ext import CodeBlockOverride, LiteralIncludeOverride\nfrom docutils.nodes import Node, admonition, literal_block, title\nfrom docutils.parsers.rst import directives\nfrom sphinx.addnodes import highlightlang\n\nfrom litestar import Litestar\n\nif TYPE_CHECKING:\n    from sphinx.application import Sphinx\n\n\nRGX_RUN = re.compile(r\"# +?run:(.*)\")\n\nAVAILABLE_PORTS = list(range(9000, 9999))\n\n\nlogger = logging.getLogger(\"sphinx\")\n\n\ndef _load_app_from_path(path: Path) -> Litestar:\n    module = importlib.import_module(str(path.with_suffix(\"\")).replace(\"/\", \".\"))\n    for obj in module.__dict__.values():\n        if isinstance(obj, Litestar):\n            return obj\n    raise RuntimeError(f\"No Litestar app found in {path}\")\n\n\n@contextmanager\ndef run_app(path: Path) -> Generator[int, None, None]:\n    \"\"\"Run an example app from a python file.\n\n    The first ``Litestar`` instance found in the file will be used as target to run.\n    \"\"\"\n    while AVAILABLE_PORTS:\n        port = AVAILABLE_PORTS.pop()\n        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:\n            if sock.connect_ex((\"127.0.0.1\", port)) != 0:\n                break\n    else:\n        raise RuntimeError(\"Could not find an open port\")\n\n    app = _load_app_from_path(path)\n\n    def run() -> None:\n        with redirect_stderr(Path(os.devnull).open()):\n            uvicorn.run(app, port=port, access_log=False)\n\n    proc = multiprocessing.Process(target=run)\n    proc.start()\n    for _ in range(50):\n        try:\n            httpx.get(f\"http://127.0.0.1:{port}\", timeout=0.1)\n        except httpx.TransportError:\n            time.sleep(0.1)\n        else:\n            break\n    try:\n        yield port\n    finally:\n        proc.kill()\n    AVAILABLE_PORTS.append(port)\n\n\ndef extract_run_args(content: str) -> tuple[str, list[list[str]]]:\n    \"\"\"Extract run args from a python file.\n\n    Return the file content stripped of the run comments and a list of argument lists\n    \"\"\"\n    new_lines = []\n    run_configs = []\n    for line in content.splitlines():\n        if run_stmt_match := RGX_RUN.match(line):\n            run_stmt = run_stmt_match.group(1).lstrip()\n            run_configs.append(shlex.split(run_stmt))\n        else:\n            new_lines.append(line)\n    return \"\\n\".join(new_lines), run_configs\n\n\ndef exec_examples(app_file: Path, run_configs: list[list[str]]) -> str:\n    \"\"\"Start a server with the example application, run the specified requests against it\n    and return their results\n    \"\"\"\n\n    results = []\n\n    with run_app(app_file) as port:\n        for run_args in run_configs:\n            url_path, *options = run_args\n            args = [\"curl\", \"-s\", f\"http://127.0.0.1:{port}{url_path}\", *options]\n            clean_args = [\"curl\", f\"http://127.0.0.1:8000{url_path}\", *options]\n\n            proc = subprocess.run(\n                args,  # noqa: S603\n                capture_output=True,\n                text=True,\n            )\n            stdout = proc.stdout.splitlines()\n            if not stdout:\n                logger.error(f\"Example: {app_file}:{args} yielded no results\")\n                continue\n\n            result = \"\\n\".join(line for line in (\"> \" + (\" \".join(clean_args)), *stdout))\n            results.append(result)\n\n    return \"\\n\".join(results)\n\n\nclass LiteralInclude(LiteralIncludeOverride):\n    option_spec = {**LiteralIncludeOverride.option_spec, \"no-run\": directives.flag}\n\n    def run(self) -> list[Node]:\n        cwd = Path.cwd()\n        docs_dir = cwd / \"docs\"\n        language = self.options.get(\"language\")\n        file = Path(self.env.relfn2path(self.arguments[0])[1])\n\n        if (language != \"python\" and file.suffix != \".py\") or \"no-run\" in self.options:\n            return super().run()\n\n        content = file.read_text()\n        clean_content, run_args = extract_run_args(content)\n\n        if not run_args:\n            return super().run()\n\n        tmp_file = self.env.tmp_examples_path / str(file.relative_to(docs_dir)).replace(\"/\", \"_\")\n\n        self.arguments[0] = \"/\" + str(tmp_file.relative_to(docs_dir))\n        tmp_file.write_text(clean_content)\n\n        nodes = super().run()\n\n        result = exec_examples(file.relative_to(cwd), run_args)\n\n        nodes.append(\n            admonition(\n                \"\",\n                title(\"\", \"Run it\"),\n                highlightlang(\n                    \"\",\n                    literal_block(\"\", result),\n                    lang=\"shell\",\n                    force=False,\n                    linenothreshold=sys.maxsize,\n                ),\n                literal_block(\"\", result),\n            )\n        )\n\n        return nodes\n\n\ndef setup(app: Sphinx) -> dict[str, bool]:\n    app.add_directive(\"literalinclude\", LiteralInclude, override=True)\n    app.add_directive(\"code-block\", CodeBlockOverride, override=True)\n\n    return {\"parallel_read_safe\": True, \"parallel_write_safe\": True}\n", "path": "tools/sphinx_ext/run_examples.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport argparse\nimport importlib.metadata\nimport json\nimport os\nimport shutil\nimport subprocess\nfrom contextlib import contextmanager\nfrom pathlib import Path\nfrom typing import TypedDict\n\nREDIRECT_TEMPLATE = \"\"\"\n<!DOCTYPE HTML>\n<html lang=\"en-US\">\n    <head>\n        <title>Page Redirection</title>\n        <meta charset=\"UTF-8\">\n        <meta http-equiv=\"refresh\" content=\"0; url={target}\">\n        <script type=\"text/javascript\">window.location.href = \"{target}\"</script>\n    </head>\n    <body>\n        You are being redirected. If this does not work, click <a href='{target}'>this link</a>\n    </body>\n</html>\n\"\"\"\n\nparser = argparse.ArgumentParser()\nparser.add_argument(\"--version\", required=False)\nparser.add_argument(\"--ignore-missing-examples-output\", action=\"store_true\", default=False)\nparser.add_argument(\"output\")\n\n\nclass VersionSpec(TypedDict):\n    versions: list[str]\n    latest: str\n\n\n@contextmanager\ndef checkout(branch: str) -> None:\n    subprocess.run([\"git\", \"checkout\", branch], check=True)  # noqa: S603 S607\n    yield\n    subprocess.run([\"git\", \"checkout\", \"-\"], check=True)  # noqa: S603 S607\n\n\ndef load_version_spec() -> VersionSpec:\n    versions_file = Path(\"docs/_static/versions.json\")\n    if versions_file.exists():\n        return json.loads(versions_file.read_text())\n    return {\"versions\": [], \"latest\": \"\"}\n\n\ndef build(output_dir: str, version: str | None, ignore_missing_output: bool) -> None:\n    if version is None:\n        version = importlib.metadata.version(\"litestar\").rsplit(\".\")[0]\n    else:\n        os.environ[\"_LITESTAR_DOCS_BUILD_VERSION\"] = version\n\n    if ignore_missing_output:\n        os.environ[\"_LITESTAR_DOCS_IGNORE_MISSING_EXAMPLE_OUTPUT\"] = \"1\"\n\n    subprocess.run([\"make\", \"docs\"], check=True)  # noqa: S603 S607\n\n    output_dir = Path(output_dir)\n    output_dir.mkdir()\n    output_dir.joinpath(\".nojekyll\").touch(exist_ok=True)\n\n    version_spec = load_version_spec()\n    is_latest = version == version_spec[\"latest\"]\n\n    docs_src_path = Path(\"docs/_build/html\")\n\n    output_dir.joinpath(\"index.html\").write_text(REDIRECT_TEMPLATE.format(target=\"latest\"))\n\n    if is_latest:\n        shutil.copytree(docs_src_path, output_dir / \"latest\", dirs_exist_ok=True)\n    shutil.copytree(docs_src_path, output_dir / version, dirs_exist_ok=True)\n\n    # copy existing versions into our output dir to preserve them when cleaning the branch\n    with checkout(\"gh-pages\"):\n        for other_version in [*version_spec[\"versions\"], \"latest\"]:\n            other_version_path = Path(other_version)\n            other_version_target_path = output_dir / other_version\n            if other_version_path.exists() and not other_version_target_path.exists():\n                shutil.copytree(other_version_path, other_version_target_path)\n\n\ndef main() -> None:\n    args = parser.parse_args()\n    build(\n        output_dir=args.output,\n        version=args.version,\n        ignore_missing_output=args.ignore_missing_output,\n    )\n\n\nif __name__ == \"__main__\":\n    main()\n", "path": "tools/build_docs.py"}, {"content": "from __future__ import annotations\n\nimport importlib\nimport logging\nimport multiprocessing\nimport os\nimport re\nimport shlex\nimport socket\nimport subprocess\nimport sys\nimport time\nfrom contextlib import contextmanager, redirect_stderr\nfrom pathlib import Path\nfrom typing import TYPE_CHECKING, Generator\n\nimport httpx\nimport uvicorn\nfrom auto_pytabs.sphinx_ext import CodeBlockOverride, LiteralIncludeOverride\nfrom docutils.nodes import Node, admonition, literal_block, title\nfrom docutils.parsers.rst import directives\nfrom sphinx.addnodes import highlightlang\n\nfrom litestar import Litestar\n\nif TYPE_CHECKING:\n    from sphinx.application import Sphinx\n\n\nRGX_RUN = re.compile(r\"# +?run:(.*)\")\n\nAVAILABLE_PORTS = list(range(9000, 9999))\n\n\nlogger = logging.getLogger(\"sphinx\")\n\nignore_missing_output = os.getenv(\"_LITESTAR_DOCS_IGNORE_MISSING_EXAMPLE_OUTPUT\", \"\") == \"1\"\n\n\ndef _load_app_from_path(path: Path) -> Litestar:\n    module = importlib.import_module(str(path.with_suffix(\"\")).replace(\"/\", \".\"))\n    for obj in module.__dict__.values():\n        if isinstance(obj, Litestar):\n            return obj\n    raise RuntimeError(f\"No Litestar app found in {path}\")\n\n\n@contextmanager\ndef run_app(path: Path) -> Generator[int, None, None]:\n    \"\"\"Run an example app from a python file.\n\n    The first ``Litestar`` instance found in the file will be used as target to run.\n    \"\"\"\n    while AVAILABLE_PORTS:\n        port = AVAILABLE_PORTS.pop()\n        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:\n            if sock.connect_ex((\"127.0.0.1\", port)) != 0:\n                break\n    else:\n        raise RuntimeError(\"Could not find an open port\")\n\n    app = _load_app_from_path(path)\n\n    def run() -> None:\n        with redirect_stderr(Path(os.devnull).open()):\n            uvicorn.run(app, port=port, access_log=False)\n\n    proc = multiprocessing.Process(target=run)\n    proc.start()\n    for _ in range(50):\n        try:\n            httpx.get(f\"http://127.0.0.1:{port}\", timeout=0.1)\n        except httpx.TransportError:\n            time.sleep(0.1)\n        else:\n            break\n    try:\n        yield port\n    finally:\n        proc.kill()\n    AVAILABLE_PORTS.append(port)\n\n\ndef extract_run_args(content: str) -> tuple[str, list[list[str]]]:\n    \"\"\"Extract run args from a python file.\n\n    Return the file content stripped of the run comments and a list of argument lists\n    \"\"\"\n    new_lines = []\n    run_configs = []\n    for line in content.splitlines():\n        if run_stmt_match := RGX_RUN.match(line):\n            run_stmt = run_stmt_match.group(1).lstrip()\n            run_configs.append(shlex.split(run_stmt))\n        else:\n            new_lines.append(line)\n    return \"\\n\".join(new_lines), run_configs\n\n\ndef exec_examples(app_file: Path, run_configs: list[list[str]]) -> str:\n    \"\"\"Start a server with the example application, run the specified requests against it\n    and return their results\n    \"\"\"\n\n    results = []\n\n    with run_app(app_file) as port:\n        for run_args in run_configs:\n            url_path, *options = run_args\n            args = [\"curl\", \"-s\", f\"http://127.0.0.1:{port}{url_path}\", *options]\n            clean_args = [\"curl\", f\"http://127.0.0.1:8000{url_path}\", *options]\n\n            proc = subprocess.run(\n                args,  # noqa: S603\n                capture_output=True,\n                text=True,\n            )\n            stdout = proc.stdout.splitlines()\n            if not stdout:\n                logger.debug(proc.stderr)\n                if not ignore_missing_output:\n                    logger.error(f\"Example: {app_file}:{args} yielded no results\")\n                continue\n\n            result = \"\\n\".join(line for line in (\"> \" + (\" \".join(clean_args)), *stdout))\n            results.append(result)\n\n    return \"\\n\".join(results)\n\n\nclass LiteralInclude(LiteralIncludeOverride):\n    option_spec = {**LiteralIncludeOverride.option_spec, \"no-run\": directives.flag}\n\n    def run(self) -> list[Node]:\n        cwd = Path.cwd()\n        docs_dir = cwd / \"docs\"\n        language = self.options.get(\"language\")\n        file = Path(self.env.relfn2path(self.arguments[0])[1])\n\n        if (language != \"python\" and file.suffix != \".py\") or \"no-run\" in self.options:\n            return super().run()\n\n        content = file.read_text()\n        clean_content, run_args = extract_run_args(content)\n\n        if not run_args:\n            return super().run()\n\n        tmp_file = self.env.tmp_examples_path / str(file.relative_to(docs_dir)).replace(\"/\", \"_\")\n\n        self.arguments[0] = \"/\" + str(tmp_file.relative_to(docs_dir))\n        tmp_file.write_text(clean_content)\n\n        nodes = super().run()\n\n        result = exec_examples(file.relative_to(cwd), run_args)\n\n        nodes.append(\n            admonition(\n                \"\",\n                title(\"\", \"Run it\"),\n                highlightlang(\n                    \"\",\n                    literal_block(\"\", result),\n                    lang=\"shell\",\n                    force=False,\n                    linenothreshold=sys.maxsize,\n                ),\n                literal_block(\"\", result),\n            )\n        )\n\n        return nodes\n\n\ndef setup(app: Sphinx) -> dict[str, bool]:\n    app.add_directive(\"literalinclude\", LiteralInclude, override=True)\n    app.add_directive(\"code-block\", CodeBlockOverride, override=True)\n\n    return {\"parallel_read_safe\": True, \"parallel_write_safe\": True}\n", "path": "tools/sphinx_ext/run_examples.py"}]}