replace argparse.Namespace plumbing with typed command inputs

This commit is contained in:
Matteo Rosati
2026-04-22 16:03:51 +02:00
parent 8ebab832d5
commit 2962a2e088
15 changed files with 560 additions and 115 deletions
+5 -4
View File
@@ -1,9 +1,10 @@
from argparse import Namespace
from __future__ import annotations
from chromy.command_inputs import AddDataInput
from chromy.utilities import ingest_file
def handle_add_data(args: Namespace) -> int:
records_added = ingest_file(args.collection, args.file)
print(f"Added {records_added} records to collection '{args.collection}'.")
def handle_add_data(command: AddDataInput) -> int:
records_added = ingest_file(command.collection, command.file)
print(f"Added {records_added} records to collection '{command.collection}'.")
return 0
+4 -3
View File
@@ -1,8 +1,9 @@
from argparse import Namespace
from __future__ import annotations
from chromy.chroma_functions import count_collection
from chromy.command_inputs import CountCollectionInput
def handle_count_collection(args: Namespace) -> int:
print(count_collection(args.collection))
def handle_count_collection(command: CountCollectionInput) -> int:
print(count_collection(command.collection))
return 0
+4 -3
View File
@@ -1,9 +1,10 @@
from argparse import Namespace
from __future__ import annotations
from chromy.chroma_functions import create_collection
from chromy.command_inputs import CreateCollectionInput
def handle_create_collection(args: Namespace) -> int:
collection_name = create_collection(args.collection)
def handle_create_collection(command: CreateCollectionInput) -> int:
collection_name = create_collection(command.collection)
print(f"Created collection '{collection_name}'.")
return 0
+13 -15
View File
@@ -1,40 +1,38 @@
from argparse import Namespace
from __future__ import annotations
from chromy.chroma_functions import delete_collection, delete_data
from chromy.command_inputs import DeleteCollectionInput, DeleteRecordsInput
def _parse_where_clause(where_clause: str) -> dict[str, str]:
condition, separator, value = where_clause.partition("=")
if separator == "":
raise ValueError("Invalid --where value. Expected <condition>=<value>.")
raise ValueError(
"Invalid --where value. Expected <condition>=<value>.")
condition = condition.strip()
value = value.strip()
if not condition or not value:
raise ValueError("Invalid --where value. Expected <condition>=<value>.")
raise ValueError(
"Invalid --where value. Expected <condition>=<value>.")
return {condition: value}
def handle_delete_collection(args: Namespace) -> int:
delete_collection(args.collection)
print(f"Deleted collection '{args.collection}'.")
def handle_delete_collection(command: DeleteCollectionInput) -> int:
delete_collection(command.collection)
print(f"Deleted collection '{command.collection}'.")
return 0
def handle_delete_records(args: Namespace) -> int:
try:
where = _parse_where_clause(args.where)
except ValueError as exc:
args.error_message = str(exc)
raise
deleted = delete_data(args.collection, where)
def handle_delete_records(command: DeleteRecordsInput) -> int:
where = _parse_where_clause(command.where)
deleted = delete_data(command.collection, where)
condition, value = next(iter(where.items()))
print(
f"Deleted {deleted} record(s) from collection '{args.collection}' "
f"Deleted {deleted} record(s) from collection '{command.collection}' "
f"where {condition}={value}."
)
return 0
+3 -2
View File
@@ -1,10 +1,11 @@
from argparse import Namespace
from __future__ import annotations
from chromy.command_inputs import ListCollectionsInput
from chromy.chroma_functions import list_collections
from chromy.utilities import print_lines
def handle_list_collections(_: Namespace) -> int:
def handle_list_collections(_: ListCollectionsInput) -> int:
collections = list_collections()
if not collections:
print("No collections found.")
+4 -3
View File
@@ -1,9 +1,10 @@
from argparse import Namespace
from __future__ import annotations
from chromy.command_inputs import QueryInput
from chromy.utilities import format_query_result, print_lines, run_query
def handle_query(args: Namespace) -> int:
result = run_query(args.collection, args.query_text)
def handle_query(command: QueryInput) -> int:
result = run_query(command.collection, command.query_text)
print_lines(format_query_result(result))
return 0