Skip to content

stream.feeds

stream.feeds

Built-in feed adapters and the feed factory.

The default feed is a watched drop directory: drop files in, they become documents. Custom feeds (an API poller, a Snowflake/warehouse query, a webhook sink) register via the angelo.stream.feeds entry-point group and are constructed by :func:build_feed the same way.

DropDirFeed

Watch a directory; yield each matching file as a Document.

The feed only discovers files; deletion/archiving of the original happens downstream (after successful ingest) per the configured archive policy. A per-process seen set avoids re-yielding the same path within one run; the daemon also dedups by document identity across runs.

Source code in stream/feeds.py 
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
class DropDirFeed:
    """Watch a directory; yield each matching file as a Document.

    The feed only discovers files; deletion/archiving of the original happens
    downstream (after successful ingest) per the configured archive policy. A
    per-process ``seen`` set avoids re-yielding the same path within one run; the
    daemon also dedups by document identity across runs.
    """

    def __init__(
        self,
        path: str = "./inbox",
        glob: str = "*",
        name: str = "drop",
        recursive: bool = False,
    ) -> None:
        self.name = name
        self.path = Path(path)
        self.glob = glob
        self.recursive = recursive
        self._seen: set[str] = set()

    def poll(self) -> Iterable[Document]:
        if not self.path.exists():
            return []
        it = self.path.rglob(self.glob) if self.recursive else self.path.glob(self.glob)
        docs: list[Document] = []
        for fp in sorted(it):
            if not fp.is_file():
                continue
            key = str(fp.resolve())
            if key in self._seen:
                continue
            self._seen.add(key)
            stat = fp.stat()
            docs.append(
                Document(
                    name=_slug(fp.stem),
                    path=str(fp),
                    title=fp.stem,
                    doc_type=fp.suffix.lstrip(".").lower(),
                    metadata={
                        "filename": fp.name,
                        "mtime": stat.st_mtime,
                        "size": stat.st_size,
                        "source_feed": self.name,
                    },
                )
            )
        return docs

build_feed 

build_feed(cfg: FeedConfig)

Construct a feed adapter from its config (built-in or entry point).

Source code in stream/feeds.py 
 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
def build_feed(cfg: FeedConfig):
    """Construct a feed adapter from its config (built-in or entry point)."""
    if cfg.type in ("drop_dir", "directory", "drop"):
        opts = cfg.options
        return DropDirFeed(
            path=str(opts.get("path", "./inbox")),
            glob=str(opts.get("glob", "*")),
            name=cfg.name,
            recursive=bool(opts.get("recursive", False)),
        )

    if cfg.type == "replay":
        from .live import ReplayFeed

        opts = cfg.options
        return ReplayFeed(
            path=str(opts.get("path", "")),
            name=cfg.name,
            stream_key=str(opts.get("stream_key", "")),
            title=str(opts.get("title", "")),
            chunk_size=int(opts.get("chunk_size", 800)),
            chunk_interval=float(opts.get("chunk_interval", 0.0)),
            metadata=dict(opts.get("metadata", {}) or {}),
            drop_dir=str(opts.get("drop_dir", "./inbox/replay")),
            flush_threshold=int(opts.get("flush_threshold", 1500)),
            overlap=int(opts.get("overlap", 400)),
            doc_type=str(opts.get("doc_type", "txt")),
        )

    from . import contrib

    factory = contrib.discover_feeds().get(cfg.type)
    if factory is None:
        raise ValueError(
            f"Unknown feed type {cfg.type!r}. Built-in: drop_dir. "
            f"Register custom feeds via the 'angelo.stream.feeds' entry point."
        )
    return factory(cfg)