MCPcopy Create free account
hub / github.com/pathwaycom/pathway / run

Method run

python/pathway/io/pyfilesystem/__init__.py:39–90  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

37 self.stored_modify_times = {}
38
39 def run(self):
40 with optional_imports("pyfilesystem"):
41 from fs.errors import ResourceNotFound as FSResourceNotFound
42
43 while True:
44 start_time = time.time()
45
46 update = self._get_snapshot_update()
47 for changed_path in update.changed_paths:
48 try:
49 with self.source.open(changed_path) as file:
50 data = file.read()
51 except FileNotFoundError:
52 logging.exception(
53 f"Failed to read file from {changed_path}. "
54 "Most likely it was deleted between the change "
55 "tracking and file read"
56 )
57 update.deleted_paths.append(changed_path)
58 continue
59
60 data = data.encode("utf-8")
61
62 provided_metadata = None
63 if self.with_metadata:
64 try:
65 metadata = self.source.getinfo(
66 changed_path,
67 namespaces=[
68 "basic",
69 "details",
70 "access",
71 ],
72 )
73 metadata_dict = self._metadata_to_dict(changed_path, metadata)
74 provided_metadata = json.dumps(metadata_dict).encode("utf-8")
75 except FSResourceNotFound:
76 logging.exception(
77 f"Failed to acquire metadata for the object: {changed_path}"
78 )
79
80 self._add(api.ref_scalar(changed_path), data, provided_metadata)
81 for deleted_path in update.deleted_paths:
82 self._remove(api.ref_scalar(deleted_path), b"")
83 self.stored_modify_times.pop(deleted_path)
84 self.commit()
85 if self.mode == STATIC_MODE_NAME:
86 break
87
88 elapsed_time = time.time() - start_time
89 if elapsed_time < self.refresh_interval:
90 time.sleep(self.refresh_interval - elapsed_time)
91
92 @property
93 def _with_metadata(self):

Callers

nothing calls this directly

Calls 9

_get_snapshot_updateMethod · 0.95
_metadata_to_dictMethod · 0.95
optional_importsFunction · 0.90
openMethod · 0.80
encodeMethod · 0.80
_addMethod · 0.80
_removeMethod · 0.80
commitMethod · 0.80
dumpsMethod · 0.45

Tested by

no test coverage detected