lsst.dax.apdb.scripts.replication Namespace Reference
Functions | |
| None | replication_list_chunks (str apdb_config) |
| int | replication_delete_chunks (str apdb_config, int chunk_id, bool force, bool print_only) |
| None | _print_chunks (Collection[ReplicaChunk] chunks) |
Function Documentation
◆ _print_chunks()
|
protected |
Print the list of chunks.
Parameters
----------
chunks : `~collections.abc.Collection` [`ReplicaChunk`]
Chunks to print.
Definition at line 94 of file replication.py.
94def _print_chunks(chunks: Collection[ReplicaChunk]) -> None:
95 """Print the list of chunks.
96
97 Parameters
98 ----------
99 chunks : `~collections.abc.Collection` [`ReplicaChunk`]
100 Chunks to print.
101 """
102 print(" Chunk Id Update time Unique Id")
103 sep = "-" * 77
104 print(sep)
105 for chunk in chunks:
106 insert_time = chunk.last_update_time
107 print(f"{chunk.id:10d} {insert_time.tai.isot}/tai {chunk.unique_id}")
108 print(sep)
109 print(f"Total: {len(chunks)}")
◆ replication_delete_chunks()
| int lsst.dax.apdb.scripts.replication.replication_delete_chunks | ( | str | apdb_config, |
| int | chunk_id, | ||
| bool | force, | ||
| bool | print_only ) |
Delete replication chunks from APDB.
Parameters
----------
apdb_config : `str`
URL for APDB configuration file.
chunk_id : `int`
Chunk to delete, all earlier chunks are deleted as well.
force : `bool`
If `True` do not ask confirmation.
print_only : `bool`
If `True` print the list of chunks, but do not delete anything.
Definition at line 49 of file replication.py.
49def replication_delete_chunks(apdb_config: str, chunk_id: int, force: bool, print_only: bool) -> int:
50 """Delete replication chunks from APDB.
51
52 Parameters
53 ----------
54 apdb_config : `str`
55 URL for APDB configuration file.
56 chunk_id : `int`
57 Chunk to delete, all earlier chunks are deleted as well.
58 force : `bool`
59 If `True` do not ask confirmation.
60 print_only : `bool`
61 If `True` print the list of chunks, but do not delete anything.
62 """
63 apdb = ApdbReplica.from_uri(apdb_config)
64 chunks = apdb.getReplicaChunks()
65 if chunks is None:
66 print("APDB instance does not support replications")
67 else:
68 chunks = sorted(chunks, key=lambda chunk: chunk.id)
69 # Chunks to delete.
70 chunks = [chunk for chunk in chunks if chunk.id <= chunk_id]
71 # Check that given chunk ID actually exists.
72 if not chunks or chunks[-1].id != chunk_id:
73 print(f"ERROR: Replication chunk with ID={chunk_id} does not exist", file=sys.stderr)
74 return 1
75
76 if print_only:
77 print("Following chunks will be deleted:")
78 _print_chunks(chunks)
79 return 0
80
81 if not force:
82 try:
83 response = input(f"{len(chunks)} chunks will be removed, y[n]? ")
84 except EOFError:
85 response = ""
86 if response not in ("y", "Y"):
87 return 0
88
89 apdb.deleteReplicaChunks(chunk.id for chunk in chunks)
90
91 return 0
92
93
◆ replication_list_chunks()
| None lsst.dax.apdb.scripts.replication.replication_list_chunks | ( | str | apdb_config | ) |
Print full list of replica chunks existing in APDB.
Parameters
----------
apdb_config : `str`
URL for APDB configuration file.
Definition at line 32 of file replication.py.
32def replication_list_chunks(apdb_config: str) -> None:
33 """Print full list of replica chunks existing in APDB.
34
35 Parameters
36 ----------
37 apdb_config : `str`
38 URL for APDB configuration file.
39 """
40 apdb = ApdbReplica.from_uri(apdb_config)
41 chunks = apdb.getReplicaChunks()
42 if chunks is not None:
43 chunks = sorted(chunks, key=lambda chunk: chunk.id)
44 _print_chunks(chunks)
45 else:
46 print("APDB instance does not support replication")
47
48
Generated on Thu Nov 27 2025 09:15:10 for LSST Applications by
1.13.2