# Local filesystem connector

> **CocoIndex v1.** This page documents CocoIndex **v1** — a ground-up redesign from v0. When writing code, ignore any v0 flow-builder DSL or deprecated decorators.
>
> Source: https://cocoindex.io/docs/connectors/localfs/ · Docs index: https://cocoindex.io/docs/llms.txt · Agent skill: https://cocoindex.io/docs/skill.md
>
> v0→v1 quick map — if you reach for these v0 symbols, stop and use the v1 form: `@cocoindex.flow_def`/`FlowBuilder` → `coco.App` + a `@coco.fn` main function; `add_collector()`/`collect()`/`export()` → declare target states (`declare_row`, `declare_file`); `cocoindex.sources/functions/targets.*` → connector APIs (`localfs.walk_dir`, `coco.ops.*`, `postgres.declare_table_target`). Full mapping + API reference: https://cocoindex.io/docs/skill.md.

The `localfs` connector provides utilities for reading files from and writing files to the local file system.

```python
from cocoindex.connectors import localfs
```

## Stable memoization with FilePath

A key feature of the `localfs` connector is [**stable memoization**](/docs/programming_guide/function) through `FilePath`. When you move your entire project directory, memoization keys remain stable as long as you use the same `ContextKey` key string for the base directory.

### Using ContextKey for stable base directories

Define a `ContextKey[pathlib.Path]` with a stable string identifier. Provide the actual path in your app's lifespan, then pass the key directly to `walk_dir()`, `declare_dir_target()`, and related functions.

```python
import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs

# Define a stable key (the string "source_dir" is the stable memo identifier)
SOURCE_DIR = coco.ContextKey[pathlib.Path]("source_dir")

@coco.fn
async def app_main() -> None:
    async for file in localfs.walk_dir(SOURCE_DIR, recursive=True):
        # file.file_path has a stable memo key based on "source_dir"
        await process(file)

# In your lifespan, provide the actual path:
async def lifespan(builder: coco.EnvBuilder) -> None:
    builder.provide(SOURCE_DIR, pathlib.Path("./data"))
```

When you move your project to a different location, just update the path in `builder.provide()` — memoization keys remain the same because they're based on the stable key string (`"source_dir"`), not the filesystem path.

### FilePath

`FilePath` combines an optional base directory key and a relative path. Passing a `ContextKey[Path]` directly to any localfs function is equivalent to constructing `FilePath(base_dir=key)`.

`FilePath` supports all `pathlib.PurePath` operations:

```python
SOURCE_DIR = coco.ContextKey[pathlib.Path]("source_dir")
source = localfs.FilePath(base_dir=SOURCE_DIR)

# Create paths using the / operator
config_path = source / "config" / "settings.json"

# Access path properties
print(config_path.name)      # "settings.json"
print(config_path.suffix)    # ".json"
print(config_path.parent)    # FilePath pointing to "config/"

# Resolve to absolute path (requires active component context)
abs_path = config_path.resolve()  # pathlib.Path
```

See [FilePath](/docs/common_resources/data_types#filepath) in Resource Types for full details.

## As source

Use `walk_dir()` to iterate over files in a directory. It returns a `DirWalker` that supports both synchronous and asynchronous iteration.

```python
def walk_dir(
    path: FilePath | Path | ContextKey[Path],
    *,
    live: bool = False,
    recursive: bool = False,
    path_matcher: FilePathMatcher | None = None,
) -> DirWalker
```

**Parameters:**

- `path` — The root directory path to walk through. Can be a `FilePath`, a `pathlib.Path`, or a `ContextKey[Path]` (equivalent to `FilePath(base_dir=path)`).
- `live` — If `True`, `items()` returns a [`LiveMapView`](/docs/advanced_topics/live_component#livemapfeed-and-livemapview) that supports live file watching via `mount_each()`.
- `recursive` — If `True`, recursively walk subdirectories.
- `path_matcher` — Optional filter for files and directories. See [PatternFilePathMatcher](/docs/common_resources/data_types#patternfilepathmatcher).

**Returns:** A `DirWalker` that supports async iteration via `async for`.

### Iterating files

`walk_dir()` returns a `DirWalker` that supports async iteration, yielding `File` objects (implementing the [`FileLike`](/docs/common_resources/data_types#filelike) base class):

```python
async for file in localfs.walk_dir("/path/to/documents", recursive=True):
    text = await file.read_text()
    ...
```

File I/O runs in a thread pool, keeping the event loop responsive.

### Keyed iteration with `items()`

`DirWalker.items()` yields keyed `(str, File)` pairs, useful for associating each file with a stable string key (its relative path):

```python
async for key, file in localfs.walk_dir("/path/to/dir", recursive=True).items():
    content = await file.read()
```

### Filtering files

Use `PatternFilePathMatcher` to filter which files and directories are included:

```python
from cocoindex.connectors import localfs
from cocoindex.resources.file import PatternFilePathMatcher

# Include only .py and .md files, exclude hidden directories and test files
matcher = PatternFilePathMatcher(
    included_patterns=["**/*.py", "**/*.md"],
    excluded_patterns=["**/.*", "**/test_*", "**/__pycache__"],
)

async for file in localfs.walk_dir("/path/to/project", recursive=True, path_matcher=matcher):
    await process(file)
```

### Live file watching

When `live=True`, `items()` returns a [`LiveMapView`](/docs/advanced_topics/live_component#livemapfeed-and-livemapview) instead of a plain `AsyncIterable`. Combined with [`mount_each()`](/docs/programming_guide/processing_component#mount_each), this enables automatic incremental file watching — new, modified, and deleted files are processed without a full rescan:

```python
files = localfs.walk_dir(
    sourcedir, recursive=True,
    path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
    live=True,
)
await coco.mount_each(process_file, files.items(), target)
```

See [Live Mode](/docs/programming_guide/live_mode) for how this works and how to enable it on the app.

### Example

```python
import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs
from cocoindex.resources.file import FileLike, PatternFilePathMatcher

SOURCE_DIR = coco.ContextKey[pathlib.Path]("source_dir")

@coco.fn
async def app_main() -> None:
    matcher = PatternFilePathMatcher(included_patterns=["**/*.md"])

    async for file in localfs.walk_dir(SOURCE_DIR, recursive=True, path_matcher=matcher):
        await coco.mount(
            coco.component_subpath("file", str(file.file_path.path)),
            process_file,
            file,
        )

@coco.fn(memo=True)
async def process_file(file: FileLike) -> None:
    text = await file.read_text()
    # ... process the file content ...
```

## As target

The `localfs` connector provides target state APIs for writing files. CocoIndex tracks what files should exist and automatically handles creation, updates, and deletion.

### declare_file

Declare a single file target. This is the simplest way to write a file.

```python
@coco.fn
def declare_file(
    path: FilePath | Path | ContextKey[Path],
    content: bytes | str,
    *,
    create_parent_dirs: bool = False,
) -> None
```

**Parameters:**

- `path` — The filesystem path for the file. Can be a `FilePath`, a `pathlib.Path`, or a `ContextKey[Path]`.
- `content` — The file content (bytes or str).
- `create_parent_dirs` — If `True`, create parent directories if they don't exist.

**Example:**

```python
OUTPUT_DIR = coco.ContextKey[pathlib.Path]("output_dir")

@coco.fn
def app_main() -> None:
    coco.mount(
        localfs.declare_file,
        localfs.FilePath("readme.txt", base_dir=OUTPUT_DIR),
        content="Hello, world!",
        create_parent_dirs=True,
    )
```

### declare_dir_target

Declare a directory target for writing multiple files. Returns a `DirTarget` for declaring files within.

```python
@coco.fn
def declare_dir_target(
    path: FilePath | Path | ContextKey[Path],
    *,
    create_parent_dirs: bool = True,
) -> DirTarget[coco.PendingS]
```

**Parameters:**

- `path` — The filesystem path for the directory. Can be a `FilePath`, a `pathlib.Path`, or a `ContextKey[Path]`.
- `create_parent_dirs` — If `True`, create parent directories if they don't exist. Defaults to `True`.

**Returns:** A pending `DirTarget`. Use `await coco.mount_target(...)` or the convenience wrapper `await localfs.mount_dir_target(path)` to resolve.

### DirTarget.declare_file

Declares a file to be written within the directory.

```python
def declare_file(
    self,
    filename: str | PurePath,
    content: bytes | str,
    *,
    create_parent_dirs: bool = False,
) -> None
```

**Parameters:**

- `filename` — The name of the file (can include subdirectory path).
- `content` — The file content (bytes or str).
- `create_parent_dirs` — If `True`, create parent directories within the target directory.

### DirTarget.declare_dir_target

Declares a subdirectory target within the directory.

```python
def declare_dir_target(
    self,
    path: str | PurePath,
    *,
    create_parent_dirs: bool = False,
) -> DirTarget[coco.PendingS]
```

**Parameters:**

- `path` — The path of the subdirectory (relative to this directory).
- `create_parent_dirs` — If `True`, create parent directories.

**Returns:** A `DirTarget` for the subdirectory.

### Target example

```python
import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs
from cocoindex.resources.file import FileLike, PatternFilePathMatcher

SOURCE_DIR = coco.ContextKey[pathlib.Path]("source_dir")
OUTPUT_DIR = coco.ContextKey[pathlib.Path]("output_dir")

@coco.fn
async def app_main() -> None:
    # Declare output directory target using context key
    target = await localfs.mount_dir_target(OUTPUT_DIR)

    # Process files and write outputs
    await coco.mount_each(process_file, localfs.walk_dir(SOURCE_DIR, recursive=True).items(), target)

@coco.fn(memo=True)
async def process_file(file: FileLike, target: localfs.DirTarget) -> None:
    # Transform the file
    content = (await file.read_text()).upper()

    # Write to output with same relative path
    target.declare_file(
        filename=file.file_path.path,
        content=content,
        create_parent_dirs=True,
    )
```