Skills › Scientific Agent Skills › omero-integration › inventory.py
inventory.py
302 lines · skills/omero-integration/scripts/inventory.py at 5f2bc03
1#!/usr/bin/env python3
2"""Produce a bounded, read-only OMERO object inventory."""
4from __future__ import annotations
6import argparse
7import sys
8from datetime import datetime, timezone
9from typing import Any
11from omero_common import (
12 ConfigError,
13 DependencyError,
14 OutputPathError,
15 bounded_int,
16 config_summary,
17 emit_json,
18 gateway_session,
19 json_safe,
20 load_connection_config,
21 scrubbed_error,
22)
25OBJECT_TYPES = (
26 "Project",
27 "Dataset",
28 "Image",
29 "Screen",
30 "Plate",
31 "Well",
32 "Fileset",
33 "OriginalFile",
34)
37def positive_argument(value: str) -> int:
38 try:
39 parsed = int(value, 10)
40 except ValueError as exc:
41 raise argparse.ArgumentTypeError("expected an integer") from exc
42 if parsed <= 0:
43 raise argparse.ArgumentTypeError("expected a positive integer")
44 return parsed
47def build_parser() -> argparse.ArgumentParser:
48 parser = argparse.ArgumentParser(
49 description=(
50 "Plan or execute a bounded read-only OMERO object inventory. "
51 "Without --execute, no server connection is made."
52 )
53 )
54 parser.add_argument(
55 "--object-type",
56 choices=OBJECT_TYPES,
57 default="Image",
58 help="Allowlisted OMERO object type (default: Image).",
59 )
60 parser.add_argument(
61 "--limit",
62 type=positive_argument,
63 default=100,
64 help="Overall object cap, 1..1000 (default: 100).",
65 )
66 parser.add_argument(
67 "--page-size",
68 type=positive_argument,
69 default=50,
70 help="Server page size, 1..200 and no greater than limit.",
71 )
72 parser.add_argument(
73 "--group-id",
74 type=positive_argument,
75 help="Optional explicit group ID; cross-group -1 is not supported.",
76 )
77 parser.add_argument(
78 "--include-names",
79 action="store_true",
80 help="Include object names; names are redacted by default.",
81 )
82 parser.add_argument(
83 "--output",
84 help="Optional .json output in an existing directory.",
85 )
86 parser.add_argument(
87 "--overwrite",
88 action="store_true",
89 help="Replace an existing regular output file, never a symlink.",
90 )
91 parser.add_argument(
92 "--execute",
93 action="store_true",
94 help="Connect and perform the read-only inventory.",
95 )
96 parser.add_argument(
97 "--allow-insecure-transport",
98 action="store_true",
99 help=(
100 "Permit OMERO_SECURE=false after explicit policy review. "
101 "Encrypted transport remains the default."
102 ),
103 )
104 return parser
107def call_or_none(obj: Any, method: str) -> Any:
108 function = getattr(obj, method, None)
109 if not callable(function):
110 return None
111 return function()
114def object_record(
115 obj: Any,
116 *,
117 requested_type: str,
118 include_names: bool,
119) -> dict[str, Any]:
120 """Serialize only a small allowlisted metadata subset."""
122 record: dict[str, Any] = {
123 "type": getattr(obj, "OMERO_CLASS", requested_type),
124 "id": call_or_none(obj, "getId"),
125 }
126 if include_names:
127 record["name"] = json_safe(
128 call_or_none(obj, "getName"),
129 max_string_length=512,
130 )
131 else:
132 record["name_redacted"] = True
134 details = call_or_none(obj, "getDetails")
135 owner = call_or_none(details, "getOwner") if details is not None else None
136 group = call_or_none(details, "getGroup") if details is not None else None
137 record["owner_id"] = call_or_none(owner, "getId")
138 record["group_id"] = call_or_none(group, "getId")
140 if requested_type == "Image":
141 record["dimensions"] = {
142 "x": call_or_none(obj, "getSizeX"),
143 "y": call_or_none(obj, "getSizeY"),
144 "z": call_or_none(obj, "getSizeZ"),
145 "c": call_or_none(obj, "getSizeC"),
146 "t": call_or_none(obj, "getSizeT"),
147 }
148 record["pixels_type"] = call_or_none(obj, "getPixelsType")
149 elif requested_type == "OriginalFile":
150 record["size_bytes"] = call_or_none(obj, "getSize")
151 record["mimetype"] = json_safe(
152 call_or_none(obj, "getMimetype"),
153 max_string_length=128,
154 )
155 return record
158def collect_inventory(
159 connection: Any,
160 *,
161 object_type: str,
162 limit: int,
163 page_size: int,
164 include_names: bool,
165) -> dict[str, Any]:
166 """Collect records with explicit server and client limits."""
168 records: list[dict[str, Any]] = []
169 offset = 0
170 exhausted = False
172 while len(records) < limit:
173 requested = min(page_size, limit - len(records))
174 page = list(
175 connection.getObjects(
176 object_type,
177 opts={
178 "limit": requested,
179 "offset": offset,
180 "order_by": "obj.id",
181 },
182 )
183 )
184 if not page:
185 exhausted = True
186 break
188 records.extend(
189 object_record(
190 obj,
191 requested_type=object_type,
192 include_names=include_names,
193 )
194 for obj in page[:requested]
195 )
196 if len(page) < requested:
197 exhausted = True
198 break
199 offset += len(page)
201 return {
202 "records": records,
203 "returned": len(records),
204 "limit_reached": len(records) == limit,
205 "server_result_exhausted": exhausted,
206 }
209def dry_run_payload(
210 args: argparse.Namespace,
211 config: Any,
212) -> dict[str, Any]:
213 return {
214 "mode": "dry-run",
215 "server_contacted": False,
216 "configuration": config_summary(config),
217 "scope": {
218 "object_type": args.object_type,
219 "group_id": args.group_id,
220 "limit": args.limit,
221 "page_size": args.page_size,
222 "include_names": args.include_names,
223 "cross_group": False,
224 },
225 "next_step": "Review scope, then add --execute for a read-only query.",
226 }
229def main(argv: list[str] | None = None) -> int:
230 parser = build_parser()
231 args = parser.parse_args(argv)
232 try:
233 bounded_int(args.limit, name="limit", minimum=1, maximum=1000)
234 bounded_int(
235 args.page_size,
236 name="page-size",
237 minimum=1,
238 maximum=200,
239 )
240 if args.page_size > args.limit:
241 raise ValueError("page-size must not exceed limit")
243 config = load_connection_config(require_auth=args.execute)
244 if not args.execute:
245 emit_json(
246 dry_run_payload(args, config),
247 output=args.output,
248 overwrite=args.overwrite,
249 )
250 return 0
252 with gateway_session(
253 config,
254 allow_insecure_transport=args.allow_insecure_transport,
255 ) as connection:
256 if args.group_id is not None:
257 connection.SERVICE_OPTS.setOmeroGroup(str(args.group_id))
258 result = collect_inventory(
259 connection,
260 object_type=args.object_type,
261 limit=args.limit,
262 page_size=args.page_size,
263 include_names=args.include_names,
264 )
266 payload = {
267 "mode": "executed-read-only",
268 "generated_at": datetime.now(timezone.utc).isoformat(),
269 "configuration": config_summary(config),
270 "scope": {
271 "object_type": args.object_type,
272 "group_id": args.group_id,
273 "limit": args.limit,
274 "page_size": args.page_size,
275 "include_names": args.include_names,
276 "cross_group": False,
277 },
278 **result,
279 }
280 emit_json(
281 payload,
282 output=args.output,
283 overwrite=args.overwrite,
284 )
285 return 0
286 except (
287 ConfigError,
288 DependencyError,
289 OutputPathError,
290 FileExistsError,
291 ValueError,
292 RuntimeError,
293 ) as error:
294 print(scrubbed_error(error), file=sys.stderr)
295 return 2
296 except Exception as error:
297 print(scrubbed_error(error), file=sys.stderr)
298 return 1
301if __name__ == "__main__":
302 raise SystemExit(main())