openai4s/server/artifacts.py

_metadata_ctime_can_lag · function · L60-L77 — def _metadata_ctime_can_lag() -> bool
_shared_artifact_writer_lock · function · L84-L93 — def _shared_artifact_writer_lock(data_dir: Path) -> threading.RLock
ArtifactOperationError · class · L138-L144 — class ArtifactOperationError(Exception)
init · method · L141-L144 — def init(self, code: int, message: str) -> None
ArtifactSession · class · L147-L150 — class ArtifactSession(Protocol)
PromotionTarget · class · L158-L168 — class PromotionTarget
FrozenCaptureSnapshot · class · L172-L177 — class FrozenCaptureSnapshot
_DelegatedCaptureToken · class · L181-L184 — class _DelegatedCaptureToken
_DelegatedCaptureClaim · class · L188-L192 — class _DelegatedCaptureClaim
_secure_upload_directory_flags · function · L195-L208 — def _secure_upload_directory_flags() -> int
_require_secure_upload_dirfd · function · L211-L217 — def _require_secure_upload_dirfd() -> None
_upload_inode · function · L220-L221 — def _upload_inode(metadata: os.stat_result) -> tuple[int, int]
_PinnedUploadDirectory · class · L224-L394 — class _PinnedUploadDirectory
init · method · L235-L249 — def init( self, descriptor: int, *, root: Path, parts: tuple[str, ...], ) -> None
enter · method · L251-L252 — def enter(self) -> "_PinnedUploadDirectory"
exit · method · L254-L255 — def exit(self, *_args: Any) -> None
close · method · L257-L260 — def close(self) -> None
_component · method · L263-L273 — def _component(name: str) -> str
open_under · method · L276-L327 — def open_under( cls, root: Path, parts: tuple[str, ...], *, create: bool, ) -> "_PinnedUploadDirectory"
assert_current · method · L329-L334 — def assert_current(self) -> None
_name_for · method · L336-L339 — def _name_for(self, path: Path) -> str
lstat · method · L341-L342 — def lstat(self, path: Path) -> os.stat_result
exists · method · L344-L349 — def exists(self, path: Path) -> bool
readlink · method · L351-L352 — def readlink(self, path: Path) -> str
open_read · method · L354-L362 — def open_read(self, path: Path) -> int
create_exclusive · method · L364-L374 — def create_exclusive(self, path: Path, mode: int = 0o600) -> int
replace · method · L376-L384 — def replace(self, source: Path, destination: Path) -> None: # POSIX rename replaces an existing non-directory destination and # is the dirfd-capable primitive whose support was checked above.
unlink · method · L386-L391 — def unlink(self, path: Path, *, missing_ok: bool = False) -> None
fsync · method · L393-L394 — def fsync(self) -> None
_upload_file_state · function · L397-L407 — def _upload_file_state(metadata: os.stat_result) -> tuple[int, int, int, int, int, int]
_PinnedUploadFile · class · L410-L503 — class _PinnedUploadFile
init · method · L422-L431 — def init( self, descriptor: int, *, directory: _PinnedUploadDirectory, path: Path, ) -> None
enter · method · L433-L434 — def enter(self) -> "_PinnedUploadFile"
exit · method · L436-L437 — def exit(self, *_args: Any) -> None
close · method · L439-L442 — def close(self) -> None
open_existing · method · L445-L448 — def open_existing( cls, directory: _PinnedUploadDirectory, path: Path ) -> "_PinnedUploadFile"
create · method · L451-L454 — def create( cls, directory: _PinnedUploadDirectory, path: Path ) -> "_PinnedUploadFile"
write · method · L456-L463 — def write(self, data: bytes) -> None
verified_bytes · method · L465-L503 — def verified_bytes( self, *, named_as: Path | None = None, size_bytes: int | None = None, checksum: str | None = None, ) -> bytes
_PinnedVersionStage · class · L507-L516 — class _PinnedVersionStage
close · method · L514-L516 — def close(self) -> None
_remote_receipt_input_versions · function · L519-L539 — def _remote_receipt_input_versions(source: Mapping[str, Any]) -> list[str]
artifact_receipt_map · function · L542-L587 — def artifact_receipt_map( receipts: Iterable[Mapping[str, Any]], ) -> dict[str, dict[str, Any]]
_take_delegated_artifact_receipts · function · L590-L598 — def _take_delegated_artifact_receipts(result: object) -> dict[str, dict[str, Any]]
DelegatedCellCaptureHooks · class · L601-L725 — class DelegatedCellCaptureHooks
init · method · L609-L619 — def init( self, manager: "ArtifactManager", session: ArtifactSession, producer_frame_id: str, emit: EventSink, ) -> None
before · method · L621-L623 — def before(self, _action: object) -> _DelegatedCaptureToken
after · method · L625-L641 — def after( self, action: object, token: _DelegatedCaptureToken, result: dict[str, Any] | None, ) -> None
before_native · method · L643-L646 — def before_native(self, action: object) -> _DelegatedCaptureToken
after_native · method · L648-L661 — def after_native( self, _action: object, token: _DelegatedCaptureToken, result: object, ) -> None
after_native_with_receipts · method · L663-L678 — def after_native_with_receipts( self, _action: object, token: _DelegatedCaptureToken, _result: object, receipts: list[dict[str, Any]], ) -> None
_validated_receipts_or_claim · method · L680-L693 — def _validated_receipts_or_claim( self, token: _DelegatedCaptureToken, result: object, ) -> dict[str, dict[str, Any]]
_capture · method · L695-L725 — def _capture( self, token: _DelegatedCaptureToken, *, language: str, producing_cell_id: str | None, artifact_receipts: Mapping[str, Mapping[str, Any]], ) -> None
_md_fence · function · L728-L731 — def _md_fence(body: str) -> str
_write_confined_text · function · L734-L772 — def _write_confined_text(workspace: Path, relative: Path, content: str) -> Path
_remote_generation · function · L782-L795 — def _remote_generation(environment: dict[str, Any]) -> bool
_same_interpreter · function · L798-L820 — def _same_interpreter(interpreter: Any, has_generation: bool = False) -> bool
ArtifactManager · class · L823-L3970 — class ArtifactManager: #: A generation ends when its kernel does, so this cannot grow without #: bound in practice. The ceiling is a backstop against a session that #: restarts its kernel thousands of times, not a tuning knob.
init · method · L830-L882 — def init( self, *, data_dir: Path, store: Any, workspace_for: Callable[[str], Path], broadcast: Callable[[str, dict], None], guess_content_type: Callable[[str], str], checksum: Callable[[Path], str], trusted_delivery: bool = False, recover_uploads: bool = True, allow_external_workspace_root: bool = False, ) -> None
writer_transaction · method · L885-L889 — def writer_transaction(self) -> Iterator[None]
recover_upload_journals · method · L891-L895 — def recover_upload_journals(self) -> None
_notify · method · L897-L904 — def _notify( self, root_frame_id: str | None, event: dict[str, Any], broadcast: Broadcast | None, ) -> None
versions_dir · method · L906-L909 — def versions_dir(self) -> Path
live_path · method · L911-L927 — def live_path(self, artifact: dict) -> Path
_upload_target · method · L929-L975 — def _upload_target( self, *, frame_id: str | None, filename: Any, exact: bool, ) -> tuple[Path, Path, str]
_upload_workspace_root · method · L977-L999 — def _upload_workspace_root(self, frame_id: str) -> Path
_exact_artifact_target · method · L1001-L1044 — def _exact_artifact_target(self, artifact: dict) -> tuple[Path, Path, str]
_open_upload_parent · method · L1047-L1061 — def _open_upload_parent( root: Path, target: Path, *, create: bool, ) -> _PinnedUploadDirectory
_open_journal_upload_parent · method · L1063-L1081 — def _open_journal_upload_parent( self, payload: Mapping[str, Any] ) -> _PinnedUploadDirectory
restore_live_path · method · L1083-L1089 — def restore_live_path(self, artifact: dict, current: dict) -> Path
_stage_version_bytes_pinned · method · L1091-L1122 — def _stage_version_bytes_pinned( self, filename: str, data: bytes ) -> _PinnedVersionStage
_promote_version_stage · method · L1125-L1147 — def _promote_version_stage( stage: _PinnedVersionStage, final: Path, *, size_bytes: int, checksum: str, ) -> None
_upload_path_exists · method · L1150-L1155 — def _upload_path_exists( path: Path, directory: _PinnedUploadDirectory | None = None ) -> bool
_remove_upload_path · method · L1158-L1169 — def _remove_upload_path( path: Path, directory: _PinnedUploadDirectory | None = None ) -> None
_fsync_directory · method · L1172-L1187 — def _fsync_directory( path: Path, directory: _PinnedUploadDirectory | None = None ) -> None
_write_durable_upload_file · method · L1190-L1217 — def _write_durable_upload_file( path: Path, data: bytes, *, directory: _PinnedUploadDirectory | None = None, ) -> None
_read_upload_journal · method · L1220-L1252 — def _read_upload_journal(journal: Path) -> dict[str, Any]
_upload_file_matches · method · L1255-L1301 — def _upload_file_matches( path: Path, size_bytes: int, checksum: str, directory: _PinnedUploadDirectory | None = None, ) -> bool
_describe_upload_live · method · L1304-L1379 — def _describe_upload_live( path: Path, directory: _PinnedUploadDirectory | None = None, *, reject_aliases: bool = False, ) -> dict[str, Any]
_read_exact_upload_live · method · L1382-L1386 — def _read_exact_upload_live(path: Path, directory: _PinnedUploadDirectory) -> bytes
_write_upload_journal · method · L1388-L1419 — def _write_upload_journal(self, journal: Path, payload: dict[str, Any]) -> None
_validated_upload_journal · method · L1421-L1559 — def _validated_upload_journal(self, journal: Path, payload: Any) -> dict[str, Any]
_upload_path_matches_previous · method · L1561-L1591 — def _upload_path_matches_previous( self, path: Path, payload: dict[str, Any], directory: _PinnedUploadDirectory | None = None, ) -> bool
_remove_new_upload_path · method · L1593-L1613 — def _remove_new_upload_path( self, path: Path, payload: dict[str, Any], directory: _PinnedUploadDirectory | None = None, *, pinned: _PinnedUploadFile | None = None, ) -> None
_restore_upload_files · method · L1615-L1719 — def _restore_upload_files( self, payload: dict[str, Any], live_parent: _PinnedUploadDirectory, *, previous_file: _PinnedUploadFile | None = None, previous_bytes: bytes | None = None, staged_file: _PinnedUploadFile | None = None, version_stage: _PinnedVersionStage | None = None, ) -> None
_restore_previous_upload_bytes · method · L1722-L1738 — def _restore_previous_upload_bytes( target: Path, previous_bytes: bytes, live_parent: _PinnedUploadDirectory, ) -> None
_abort_upload · method · L1740-L1808 — def _abort_upload( self, payload: dict[str, Any], live_parent: _PinnedUploadDirectory, *, previous_file: _PinnedUploadFile | None = None, previous_bytes: bytes | None = None, staged_file: _PinnedUploadFile | None = None, version_stage: _PinnedVersionStage | None = None, ) -> None
_recover_upload_journal · method · L1810-L1868 — def _recover_upload_journal(self, journal: Path) -> None
_recover_upload_journals · method · L1870-L1893 — def _recover_upload_journals(self) -> None
freeze_capture_snapshot · method · L1895-L1972 — def freeze_capture_snapshot( self, filename: str, source_path: Path ) -> FrozenCaptureSnapshot
write_version_snapshot · method · L1974-L1998 — def write_version_snapshot( self, version_id: str, filename: str, *, src_path: Path | None = None, data: bytes | None = None, ) -> None
protect_latest · method · L2000-L2024 — def protect_latest(self, session: ArtifactSession) -> None
restore · method · L2026-L2030 — def restore(self, artifact_id: str, version_id: str) -> dict
materialise_version · method · L2032-L2085 — def materialise_version( self, *, source_version_id: str, filename: str, frame_id: str | None, workspace_frame_id: str, project_id: str, raw: bytes, producing_cell_id: str | None = None, ) -> dict
_restore_locked · method · L2087-L2168 — def _restore_locked(self, artifact_id: str, version_id: str) -> dict
edit · method · L2170-L2184 — def edit( self, artifact_id: str, content: str, *, broadcast: Broadcast | None = None, ) -> dict
_edit_locked · method · L2186-L2259 — def _edit_locked( self, artifact_id: str, content: str, *, broadcast: Broadcast | None = None, ) -> dict
_validated_unchanged_exact_artifact · method · L2261-L2303 — def _validated_unchanged_exact_artifact( self, artifact_id: str, *, raw: bytes, checksum: str, ) -> str | None
rename · method · L2305-L2317 — def rename( self, artifact_id: str, filename: str | None, *, broadcast: Broadcast | None = None, ) -> dict
_rename_locked · method · L2319-L2347 — def _rename_locked( self, artifact_id: str, filename: str | None, *, broadcast: Broadcast | None = None, ) -> dict
_upload_bytes · method · L2350-L2408 — def _upload_bytes(payload: dict) -> bytes
upload · method · L2410-L2419 — def upload( self, payload: dict, *, broadcast: Broadcast | None = None, ) -> dict
replace_artifact_text · method · L2421-L2453 — def replace_artifact_text( self, artifact_id: str, content: str, *, broadcast: Broadcast | None = None, ) -> dict
_upload_locked · method · L2455-L2546 — def _upload_locked( self, payload: dict, *, broadcast: Broadcast | None = None, exact_artifact: dict[str, Any] | None = None, raw_override: bytes | None = None, restore_source_version_id: str | None = None, materialise_source_version_id: str | None = None, producing_cell_id: str | None = None, ) -> dict
_upload_locked_in_parent · method · L2548-L3017 — def _upload_locked_in_parent( self, payload: dict, *, raw: bytes, target: Path, stored_filename: str, frame_id: str | None, workspace_frame_id: str | None, project_id: str | None, live_parent: _PinnedUploadDirectory, broadcast: Broadcast | None, exact_artifact: dict[str, Any] | None, restore_source_version_id: str | None, materialise_source_version_id: str | None, producing_cell_id: str | None, ) -> dict
publish · function · L2707-L2801 — def publish(version_id: str, artifact_id: str) -> str
delete · method · L3019-L3026 — def delete( self, artifact_id: str, *, broadcast: Broadcast | None = None, ) -> dict
_delete_locked · method · L3028-L3080 — def _delete_locked( self, artifact_id: str, *, broadcast: Broadcast | None = None, ) -> dict
snapshot · method · L3082-L3107 — def snapshot(self, workspace: Path) -> WorkspaceSnapshot
_content_fingerprints_required · method · L3110-L3131 — def _content_fingerprints_required() -> bool
_metadata_fingerprint · method · L3134-L3157 — def _metadata_fingerprint(path: Path) -> WorkspaceFileState | None
_live_fingerprint · method · L3160-L3205 — def _live_fingerprint(path: Path) -> WorkspaceFileState | None
_claim_key · method · L3208-L3209 — def _claim_key(path: Path | str) -> str
_claim_workspace_key · method · L3212-L3213 — def _claim_workspace_key(workspace: Path | str) -> str
_put_delegated_claim · method · L3215-L3242 — def _put_delegated_claim( self, path: Path, *, workspace: Path, failed: bool, ) -> None
claim_delegated_artifacts · method · L3244-L3254 — def claim_delegated_artifacts( self, artifacts: list[dict], *, workspace: Path ) -> None
claim_delegated_changes · method · L3256-L3275 — def claim_delegated_changes( self, workspace: Path, before: WorkspaceSnapshot, *, failed: bool, ) -> None
_matches_delegated_claim · method · L3277-L3303 — def _matches_delegated_claim( self, path: Path, *, workspace: Path, consume_success: bool, ) -> bool
delegated_cell_hooks · method · L3305-L3313 — def delegated_cell_hooks( self, session: ArtifactSession, producer_frame_id: str, emit: EventSink, ) -> DelegatedCellCaptureHooks
_prevalidate_receipt_lineage · method · L3315-L3360 — def _prevalidate_receipt_lineage( self, session: ArtifactSession, receipts: Mapping[str, Mapping[str, Any]], ) -> dict[str, list[str]]
register_file · method · L3362-L3478 — def register_file( self, session: ArtifactSession, path: Path, cell_id: str | None, emit: EventSink, env_snapshot_id: str | None = None, *, producer_frame_id: str | None = None, source: Any = None, input_version_ids: list[str] | tuple[str, ...] | None = None, expected_checksum: str | None = None, frozen_snapshot: FrozenCaptureSnapshot | None = None, ) -> dict | None
promote_cell · method · L3480-L3513 — def promote_cell( self, session: ArtifactSession, cell: dict, emit: EventSink, ) -> dict | None
_render_cell_markdown · method · L3515-L3540 — def _render_cell_markdown(self, cell: dict, workspace: Path) -> str
_render_promoted_figure · method · L3543-L3561 — def _render_promoted_figure(workspace: Path, figure: str) -> str
capture · method · L3563-L3754 — def capture( self, session: ArtifactSession, cell_index: int, cell_id: str | None, before: WorkspaceSnapshot, emit: EventSink, language: str = "python", run_system_cell: Callable[[str], dict] | None = None, drain_remote_provenance: Callable[[], Any] | None = None, *, producer_frame_id: str | None = None, honor_delegated_claims: bool = True, artifact_receipts: Mapping[str, Mapping[str, Any]] | None = None, ) -> CaptureResult
capture_environment · method · L3756-L3785 — def capture_environment( self, drain_remote_provenance: Callable[[], Any] | None = None, *, root_frame_id: str | None = None, language: str = "python", ) -> str | None
_generation_for · method · L3787-L3810 — def _generation_for( self, root_frame_id: str | None, language: str ) -> dict[str, Any] | None
_snapshot_for · method · L3812-L3918 — def _snapshot_for( self, generation: dict[str, Any] | None, language: str ) -> dict[str, Any]
invalidate_freeze_cache · method · L3920-L3932 — def invalidate_freeze_cache(self) -> None
_frozen_packages · method · L3934-L3970 — def _frozen_packages( self, interpreter: Any, generation: dict[str, Any] | None ) -> list[dict[str, Any]] | None
_capture_snippet · function · L3973-L3997 — def _capture_snippet(index: int) -> str: # pyplot is read from sys.modules, never imported, as in # kernel/guards.py: importing it builds matplotlib's font list, and an # enforced sandbox gives each kernel an empty private MPLCONFIGDIR. The # old 'matplotlib' in sys.modules test imported pyplot after a Cell that # only imported matplotlib, or whose first pyplot import was stopped, # so a font scan (8-48s on macOS) ran after the Cell and the Stop or the # daemon shutdown waited it out. A process that has not imported pyplot # has no pyplot figures to capture.
_ignored_file · function · L4000-L4009 — def _ignored_file(path: Path) -> bool
is_text_editable · function · L4012-L4021 — def is_text_editable(filename: str | None, content_type: str | None) -> bool