|
| 1 | +import dataclasses |
| 2 | +import datetime |
| 3 | + |
| 4 | +import cocoindex |
| 5 | +from dotenv import load_dotenv |
| 6 | + |
| 7 | + |
| 8 | +@dataclasses.dataclass |
| 9 | +class Document: |
| 10 | + filename: str |
| 11 | + content: str |
| 12 | + |
| 13 | + |
| 14 | +@cocoindex.flow_def(name="LiveUpdates") |
| 15 | +def live_update_flow( |
| 16 | + flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope |
| 17 | +): |
| 18 | + # Source: local files in the 'data' directory |
| 19 | + data_scope["documents"] = flow_builder.add_source( |
| 20 | + cocoindex.sources.LocalFile(path="data"), |
| 21 | + refresh_interval=datetime.timedelta(seconds=5), |
| 22 | + ) |
| 23 | + |
| 24 | + # Collector |
| 25 | + collector = data_scope.add_collector() |
| 26 | + with data_scope["documents"].row() as doc: |
| 27 | + collector.collect( |
| 28 | + filename=doc["filename"], |
| 29 | + content=doc["content"], |
| 30 | + ) |
| 31 | + |
| 32 | + # Target: Postgres database |
| 33 | + collector.export( |
| 34 | + "documents_index", |
| 35 | + cocoindex.targets.Postgres(), |
| 36 | + primary_key_fields=["filename"], |
| 37 | + ) |
| 38 | + |
| 39 | + |
| 40 | +def main(): |
| 41 | + # Setup the flow |
| 42 | + live_update_flow.setup(report_to_stdout=True) |
| 43 | + |
| 44 | + # Start the live updater |
| 45 | + print("Starting live updater...") |
| 46 | + with cocoindex.FlowLiveUpdater( |
| 47 | + live_update_flow, cocoindex.FlowLiveUpdaterOptions(print_stats=True) |
| 48 | + ) as updater: |
| 49 | + print("Live updater started. Watching for changes in the 'data' directory.") |
| 50 | + print("Try adding, modifying, or deleting files in the 'data' directory.") |
| 51 | + print("Press Ctrl+C to stop.") |
| 52 | + updater.wait() |
| 53 | + |
| 54 | + |
| 55 | +if __name__ == "__main__": |
| 56 | + load_dotenv() |
| 57 | + cocoindex.init() |
| 58 | + main() |
0 commit comments