Skip to content

resource_registry

Class info

Classes

Name Children Inherits
ResourceRegistry
llmling_agent.resource_registry
Registry for filesystem resources.

    🛈 DocStrings

    ResourceRegistry

    Bases: BaseRegistry[str, AbstractFileSystem]

    Registry for filesystem resources.

    Source code in src/llmling_agent/resource_registry.py 
     20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
    class ResourceRegistry(BaseRegistry[str, AbstractFileSystem]):
    """Registry for filesystem resources."""

    @property
    def _error_class(self) -> type[LLMLingError]:
        return LLMLingError

    def register(self, name: str, item: Any, replace: bool = False):
        """Register a new resource."""
        logger.debug("registering %r (%r)", name, item.__class__.__name__)
        fsspec.register_implementation(name, item.__class__, clobber=True)
        super().register(name, item, replace=replace)

    def _validate_item(self, item: Any) -> AbstractFileSystem:
        if not isinstance(item, AbstractFileSystem):
            msg = f"Expected AbstractFileSystem, got {type(item)}"
            raise self._error_class(msg)
        return item

    def register_from_config(
        self, name: str, config: ResourceConfig
    ) -> AbstractFileSystem:
        """Register a new resource from config."""
        from llmling_agent_config.resources import SourceResourceConfig

        match config:
            case SourceResourceConfig():
                # Extract base path from URI if present
                protocol, path = (
                    config.uri.split("://", 1)
                    if "://" in config.uri
                    else (config.uri, "")
                )
                if path:
                    config.storage_options["root"] = path

                fs = fsspec.filesystem(protocol, **config.storage_options)

                if config.path:
                    fs = fsspec.filesystem("dir", fs=fs, path=config.path)
                if config.cached:
                    fs = fsspec.filesystem("cached", fs=fs)

                self.register(name, fs)
                return fs
            case _:
                msg = f"Unknown resource config type: {type(config)}"
                raise ValueError(msg)

    def get_fs(self) -> UnionFileSystem:
        """Get unified filesystem view of all resources."""
        filesystems = dict(self.items())
        return UnionFileSystem(filesystems)

    def get_upath(self, resource_name: str | None = None) -> UPath:
        """Get a UPath object for accessing a resource."""
        return UPath("union://")
        # path._fs_cached = self.get_fs()
        # return path

    async def get_content(
        self,
        path: str,
        encoding: str = "utf-8",
        recursive: bool = True,
        exclude: list[str] | None = None,
        max_depth: int | None = None,
    ) -> str:
        """Get content from a resource as text.

        Args:
            path: Path to read, either:
                - resource (whole resource)
                - resource:// (whole resource)
                - resource://file.txt (single file)
                - resource://folder (directory)
            encoding: Text encoding for binary content
            recursive: For directories, whether to read recursively
            exclude: For directories, patterns to exclude
            max_depth: For directories, maximum depth to read

        Returns:
            For files: file content
            For directories: concatenated content of all files
        """
        from upathtools import read_folder, read_path

        if "/" not in path:
            path = f"{path}://"
        resource, _ = path.split("://", 1)
        fs = self.get_fs()
        is_dir = await fs._isdir(path)

        if is_dir:
            content_dict = await read_folder(
                path,
                mode="rt",
                encoding=encoding,
                recursive=recursive,
                exclude=exclude,
                max_depth=max_depth,
            )
            # Combine all files with headers
            sections = []
            for rel_path, content in sorted(content_dict.items()):
                sections.extend([f"--- {rel_path} ---", content, ""])
            return "\n".join(sections)
        return await read_path(path, encoding=encoding)

    async def query(
        self,
        path: str,
        pattern: str = "**/*",
        *,
        recursive: bool = True,
        include_dirs: bool = False,
        exclude: list[str] | None = None,
        max_depth: int | None = None,
    ) -> list[str]:
        """Query contents of a resource or subfolder.

        Args:
            path: Path to query, either:
                - resource (queries whole resource)
                - resource:// (queries whole resource)
                - resource://subfolder (queries specific folder)
            pattern: Glob pattern to match files against
            recursive: Whether to search subdirectories
            include_dirs: Whether to include directories in results
            exclude: List of patterns to exclude
            max_depth: Maximum directory depth for recursive search

        Example:
            # Query whole resource (all equivalent)
            files = await registry.query("docs")
            files = await registry.query("docs://")

            # Query specific subfolder
            files = await registry.query("docs://guides", pattern="*.md")
        """
        from upathtools import list_files

        if "/" not in path:
            # Simple resource name - add protocol
            path = f"{path}://"
        resource, _ = path.split("://", 1)
        if resource not in self:
            msg = f"Resource not found: {resource}"
            raise ValueError(msg)

        files = await list_files(
            path,
            pattern=pattern,
            recursive=recursive,
            include_dirs=include_dirs,
            exclude=exclude,
            max_depth=max_depth,
        )
        return [str(p) for p in files]

    get_content async 

    get_content(
    path: str,
    encoding: str = "utf-8",
    recursive: bool = True,
    exclude: list[str] | None = None,
    max_depth: int | None = None,
) -> str

    Get content from a resource as text.

    Parameters:

    Name Type Description Default
    path str

    Path to read, either: - resource (whole resource) - resource:// (whole resource) - resource://file.txt (single file) - resource://folder (directory)

    		 required
    encoding str

    Text encoding for binary content

    		 'utf-8'
    recursive bool

    For directories, whether to read recursively

    		 True
    exclude list[str] | None

    For directories, patterns to exclude

    		 None
    max_depth int | None

    For directories, maximum depth to read

    		 None

    Returns:

    Type Description
    str

    For files: file content
    str

    For directories: concatenated content of all files
    Source code in src/llmling_agent/resource_registry.py 
     80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
    async def get_content(
    self,
    path: str,
    encoding: str = "utf-8",
    recursive: bool = True,
    exclude: list[str] | None = None,
    max_depth: int | None = None,
) -> str:
    """Get content from a resource as text.

    Args:
        path: Path to read, either:
            - resource (whole resource)
            - resource:// (whole resource)
            - resource://file.txt (single file)
            - resource://folder (directory)
        encoding: Text encoding for binary content
        recursive: For directories, whether to read recursively
        exclude: For directories, patterns to exclude
        max_depth: For directories, maximum depth to read

    Returns:
        For files: file content
        For directories: concatenated content of all files
    """
    from upathtools import read_folder, read_path

    if "/" not in path:
        path = f"{path}://"
    resource, _ = path.split("://", 1)
    fs = self.get_fs()
    is_dir = await fs._isdir(path)

    if is_dir:
        content_dict = await read_folder(
            path,
            mode="rt",
            encoding=encoding,
            recursive=recursive,
            exclude=exclude,
            max_depth=max_depth,
        )
        # Combine all files with headers
        sections = []
        for rel_path, content in sorted(content_dict.items()):
            sections.extend([f"--- {rel_path} ---", content, ""])
        return "\n".join(sections)
    return await read_path(path, encoding=encoding)

    get_fs 

    get_fs() -> UnionFileSystem

    Get unified filesystem view of all resources.

    Source code in src/llmling_agent/resource_registry.py 
    69
70
71
72
    def get_fs(self) -> UnionFileSystem:
    """Get unified filesystem view of all resources."""
    filesystems = dict(self.items())
    return UnionFileSystem(filesystems)

    get_upath 

    get_upath(resource_name: str | None = None) -> UPath

    Get a UPath object for accessing a resource.

    Source code in src/llmling_agent/resource_registry.py 
    74
75
76
    def get_upath(self, resource_name: str | None = None) -> UPath:
    """Get a UPath object for accessing a resource."""
    return UPath("union://")

    query async 

    query(
    path: str,
    pattern: str = "**/*",
    *,
    recursive: bool = True,
    include_dirs: bool = False,
    exclude: list[str] | None = None,
    max_depth: int | None = None,
) -> list[str]

    Query contents of a resource or subfolder.

    Parameters:

    Name Type Description Default
    path str

    Path to query, either: - resource (queries whole resource) - resource:// (queries whole resource) - resource://subfolder (queries specific folder)

    		 required
    pattern str

    Glob pattern to match files against

    		 '**/*'
    recursive bool

    Whether to search subdirectories

    		 True
    include_dirs bool

    Whether to include directories in results

    		 False
    exclude list[str] | None

    List of patterns to exclude

    		 None
    max_depth int | None

    Maximum directory depth for recursive search

    		 None
    Example

    Query whole resource (all equivalent)

    files = await registry.query("docs") files = await registry.query("docs://")

    Query specific subfolder

    files = await registry.query("docs://guides", pattern="*.md")

    Source code in src/llmling_agent/resource_registry.py 
    129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
    async def query(
    self,
    path: str,
    pattern: str = "**/*",
    *,
    recursive: bool = True,
    include_dirs: bool = False,
    exclude: list[str] | None = None,
    max_depth: int | None = None,
) -> list[str]:
    """Query contents of a resource or subfolder.

    Args:
        path: Path to query, either:
            - resource (queries whole resource)
            - resource:// (queries whole resource)
            - resource://subfolder (queries specific folder)
        pattern: Glob pattern to match files against
        recursive: Whether to search subdirectories
        include_dirs: Whether to include directories in results
        exclude: List of patterns to exclude
        max_depth: Maximum directory depth for recursive search

    Example:
        # Query whole resource (all equivalent)
        files = await registry.query("docs")
        files = await registry.query("docs://")

        # Query specific subfolder
        files = await registry.query("docs://guides", pattern="*.md")
    """
    from upathtools import list_files

    if "/" not in path:
        # Simple resource name - add protocol
        path = f"{path}://"
    resource, _ = path.split("://", 1)
    if resource not in self:
        msg = f"Resource not found: {resource}"
        raise ValueError(msg)

    files = await list_files(
        path,
        pattern=pattern,
        recursive=recursive,
        include_dirs=include_dirs,
        exclude=exclude,
        max_depth=max_depth,
    )
    return [str(p) for p in files]

    register 

    register(name: str, item: Any, replace: bool = False)

    Register a new resource.

    Source code in src/llmling_agent/resource_registry.py 
    27
28
29
30
31
    def register(self, name: str, item: Any, replace: bool = False):
    """Register a new resource."""
    logger.debug("registering %r (%r)", name, item.__class__.__name__)
    fsspec.register_implementation(name, item.__class__, clobber=True)
    super().register(name, item, replace=replace)

    register_from_config 

    register_from_config(name: str, config: ResourceConfig) -> AbstractFileSystem

    Register a new resource from config.

    Source code in src/llmling_agent/resource_registry.py 
    39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
    def register_from_config(
    self, name: str, config: ResourceConfig
) -> AbstractFileSystem:
    """Register a new resource from config."""
    from llmling_agent_config.resources import SourceResourceConfig

    match config:
        case SourceResourceConfig():
            # Extract base path from URI if present
            protocol, path = (
                config.uri.split("://", 1)
                if "://" in config.uri
                else (config.uri, "")
            )
            if path:
                config.storage_options["root"] = path

            fs = fsspec.filesystem(protocol, **config.storage_options)

            if config.path:
                fs = fsspec.filesystem("dir", fs=fs, path=config.path)
            if config.cached:
                fs = fsspec.filesystem("cached", fs=fs)

            self.register(name, fs)
            return fs
        case _:
            msg = f"Unknown resource config type: {type(config)}"
            raise ValueError(msg)