Support route static from binary in aiohttp

commit_hash:3fd15af6d92c77e1aaf327bd7f34742551ebf3b6

1 files changed, 111 insertions, 19 deletions

diff --git a/library/python/runtime_py3/sitecustomize.py b/library/python/runtime_py3/sitecustomize.py
index 3b30b8807e8..dbd134f5a47 100644
--- a/library/python/runtime_py3/sitecustomize.py
+++ b/library/python/runtime_py3/sitecustomize.py
@@ -1,8 +1,11 @@
-import pathlib
+import functools
 import io
 import os
+import pathlib
 import re
+import stat
 import sys
+import typing
 
 from importlib.metadata import (
     Distribution,
@@ -11,14 +14,57 @@ from importlib.metadata import (
 )
 from importlib.resources.abc import Traversable
 
-from __res import _ResfsResourceReader, find, iter_keys, resfs_read, resfs_files
+from __res import find, iter_keys, resfs_files, resfs_read
 
 METADATA_NAME = re.compile("^Name: (.*)$", re.MULTILINE)
 
+try:
+    BINARY_STAT: typing.Final[os.stat_result] = os.stat(sys.executable)
+except FileNotFoundError:
+    BINARY_STAT: typing.Final[os.stat_result] = os.stat_result(
+        (
+            0,  # st_mode
+            1,  # st_ino
+            0,  # st_dev
+            1,  # st_nlink
+            0,  # st_uid
+            0,  # st_gid
+            0,  # st_size
+            0,  # st_atime
+            0,  # st_mtime
+            0,  # st_ctime
+        ),
+        {
+            "st_atime_ns": 0,
+            "st_mtime_ns": 0,
+            "st_ctime_ns": 0,
+        },
+    )
+
+RESOURCE_DIRECTORY_STAT: typing.Final[os.stat_result] = os.stat_result(
+    (
+        stat.S_IFDIR | 0o755,  # st_mode
+        1,  # st_ino
+        0,  # st_dev
+        1,  # st_nlink
+        BINARY_STAT.st_uid,  # st_uid
+        BINARY_STAT.st_gid,  # st_gid
+        0,  # st_size
+        BINARY_STAT.st_atime,  # st_atime
+        BINARY_STAT.st_mtime,  # st_mtime
+        BINARY_STAT.st_ctime,  # st_ctime
+    ),
+    {
+        "st_atime_ns": BINARY_STAT.st_atime_ns,
+        "st_mtime_ns": BINARY_STAT.st_mtime_ns,
+        "st_ctime_ns": BINARY_STAT.st_ctime_ns,
+    },
+)
+
 
 class ArcadiaTraversable(Traversable):
-    def __init__(self, resfs):
-        self._resfs = resfs
+    def __init__(self, resfs) -> None:
+        self._resfs = str(resfs)
         self._path = pathlib.Path(resfs)
 
     def __eq__(self, other) -> bool:
@@ -34,28 +80,45 @@ class ArcadiaTraversable(Traversable):
     def __hash__(self) -> int:
         return hash(self._path)
 
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({self._resfs!r})"
+
     @property
-    def name(self):
+    def name(self) -> str:
         return self._path.name
 
     @property
-    def suffix(self):
+    def suffix(self) -> str:
         return self._path.suffix
 
     @property
-    def stem(self):
+    def stem(self) -> str:
         return self._path.stem
 
+    def with_suffix(self, suffix: str) -> typing.Self:
+        return type(self)(self._path.with_suffix(suffix))
+
+    def lstat(self) -> os.stat_result:
+        return self.stat()
+
+    def relative_to(self, other):
+        if isinstance(other, ArcadiaTraversable):
+            return self._path.relative_to(other._path)
+        raise NotImplementedError
+
+    def resolve(self, strict: bool = False) -> typing.Self:
+        return self
+
 
 class ArcadiaResource(ArcadiaTraversable):
-    def is_file(self):
+    def is_file(self) -> bool:
         return True
 
-    def is_dir(self):
+    def is_dir(self) -> bool:
         return False
 
     def open(self, mode="r", *args, **kwargs):
-        data = find(self._resfs.encode("utf-8"))
+        data = self.data
         if data is None:
             raise FileNotFoundError(self._resfs)
 
@@ -66,21 +129,47 @@ class ArcadiaResource(ArcadiaTraversable):
 
         return stream
 
-    def joinpath(self, *name):
+    def joinpath(self, *name) -> ArcadiaTraversable:
         raise RuntimeError("Cannot traverse into a resource")
 
     def iterdir(self):
         return iter(())
 
-    def __repr__(self) -> str:
-        return f"ArcadiaResource({self._resfs!r})"
+    @functools.cached_property
+    def data(self) -> bytes | None:
+        return find(self._resfs.encode("utf-8"))
+
+    def stat(self) -> os.stat_result:
+        data = self.data
+        if data is None:
+            raise FileNotFoundError(self._resfs)
+
+        return os.stat_result(
+            (
+                stat.S_IFREG | 0o644,  # st_mode
+                1,  # st_ino
+                0,  # st_dev
+                1,  # st_nlink
+                BINARY_STAT.st_uid,  # st_uid
+                BINARY_STAT.st_gid,  # st_gid
+                len(data),  # st_size
+                BINARY_STAT.st_atime,  # st_atime
+                BINARY_STAT.st_mtime,  # st_mtime
+                BINARY_STAT.st_ctime,  # st_ctime
+            ),
+            {
+                "st_atime_ns": BINARY_STAT.st_atime_ns,
+                "st_mtime_ns": BINARY_STAT.st_mtime_ns,
+                "st_ctime_ns": BINARY_STAT.st_ctime_ns,
+            },
+        )
 
 
 class ArcadiaResourceContainer(ArcadiaTraversable):
-    def is_dir(self):
+    def is_dir(self) -> bool:
         return True
 
-    def is_file(self):
+    def is_file(self) -> bool:
         return False
 
     def iterdir(self):
@@ -100,14 +189,16 @@ class ArcadiaResourceContainer(ArcadiaTraversable):
     @staticmethod
     def _flatten(compound_names):
         for name in compound_names:
-            yield from name.split("/")
+            yield from str(name).split("/")
 
-    def joinpath(self, *descendants):
+    def joinpath(self, *descendants) -> ArcadiaTraversable:
         if not descendants:
             return self
 
         names = self._flatten(descendants)
         target = next(names)
+        if target == ".":
+            return self
         for traversable in self.iterdir():
             if traversable.name == target:
                 if isinstance(traversable, ArcadiaResource):
@@ -117,8 +208,9 @@ class ArcadiaResourceContainer(ArcadiaTraversable):
 
         raise FileNotFoundError("/".join(self._flatten(descendants)))
 
-    def __repr__(self):
-        return f"ArcadiaResourceContainer({self._resfs!r})"
+    @staticmethod
+    def stat() -> os.stat_result:
+        return RESOURCE_DIRECTORY_STAT
 
 
 class ArcadiaDistribution(Distribution):