PYL-26 software design improvement separate domain code from cli

src/pylms/__main__.py

@@ -1,10 +1,17 @@
 #!/bin/env python3
 
+import logging
 from sys import argv
+import pylms.pylms
 from pylms.pylms import list_persons, store_person, update_person, delete_person, link_persons, ExitPyLMS
+from pylms.cli import CLI
 
 
 def main() -> None:
+ _cli = CLI()
+ pylms.pylms.ios = _cli
+ pylms.pylms.events = _cli
+ logging.basicConfig(level=logging.INFO)
  try:
  _read_and_execute_commands()
  except ExitPyLMS:

src/pylms/cli.py

@@ -0,0 +1,111 @@
+from pylms.pylms import IOs, EventListener, ExitPyLMS
+from pylms.pylms import Person, Relationship, RelationshipDefinition
+
+
+class CLI(IOs, EventListener):
+ def creating_person(self, person: Person) -> None:
+ print(f"Create Person {person}.")
+
+ def show_person(self, person: Person) -> None:
+ created = person.created
+ print(
+ f"({person.person_id})",
+ person,
+ f"({created.year}-{created.month}-{created.day} {created.hour}-{created.minute}-{created.second})",
+ )
+
+ def list_persons(self, resolved_persons: list[(Person, list[Relationship])]) -> None:
+ if resolved_persons:
+ for person, rs in sorted(resolved_persons, key=lambda t: t[0].person_id):
+ self.show_person(person)
+ for r in rs:
+ other = r.right if r.left == person else r.left
+ print(f" -> {r.repr_for(person)} de ({other.person_id}) {other}")
+ else:
+ print("No Person registered yet.")
+
+ def _interactive_hit_enter(self):
+ while True:
+ s = self._input_or_exit_pylms()
+
+ if len(s) == 0:
+ return
+
+ print("Just hit ENTER")
+
+ def _input_or_exit_pylms(self):
+ try:
+ return input()
+ except KeyboardInterrupt:
+ raise ExitPyLMS()
+
+ def _interactive_person_id(self, valid_ids: list[int]) -> int:
+ if not valid_ids:
+ raise ValueError("valid_ids can not be empty.")
+
+ while True:
+ n = self._input_or_exit_pylms()
+
+ try:
+ res = int(n)
+ if res not in valid_ids:
+ print("Not a valid id.")
+ continue
+
+ return res
+ except ValueError:
+ print("Not an integer.")
+
+ def select_person(self, persons: list[Person]) -> Person | None:
+ print("Input id of person to update:")
+ for person in sorted(persons, key=lambda p: p.person_id):
+ self.show_person(person)
+ print("CTRL+C to exit")
+
+ person_id = self._interactive_person_id([person.person_id for person in persons])
+ for person in persons:
+ if person.person_id == person_id:
+ return person
+
+ # should not happen
+ raise RuntimeError(f"id {person_id} does not exist in list of Persons")
+
+ def update_person(self, person_to_update: Person) -> Person:
+ print("Input new first name and last name to update:")
+ self.show_person(person_to_update)
+ print("CTRL+C to exit")
+
+ firstname: str
+ lastname: str
+ firstname, lastname = self._get_person_details()
+
+ person_to_update.firstname = firstname
+ person_to_update.lastname = lastname
+
+ return person_to_update
+
+ def _get_person_details(self) -> tuple[str, str | None]:
+ while True:
+ text = self._input_or_exit_pylms()
+
+ words = text.split(" ")
+ if len(words) > 2:
+ print("Too many words.")
+ continue
+
+ if len(words) == 1:
+ return words[0], None
+ return words[0], words[1]
+
+ def deleting_person(self, person_to_delete: Person) -> None:
+ print("Hit ENTER to delete:")
+ self.show_person(person_to_delete)
+ print("CTRL+C to exit")
+ self._interactive_hit_enter()
+
+ def creating_link(self, rl_definition: RelationshipDefinition, person_left: Person, person_right: Person) -> None:
+ print(f'Hit ENTER to link as "{rl_definition.name}":')
+ self.show_person(person_left)
+ self.show_person(person_right)
+ print("CTRL+C to exit")
+ self._interactive_hit_enter()

src/pylms/pylms.py

@@ -2,122 +2,92 @@
 from pylms.core import Person, PersonIdGenerator
 from pylms.core import relationship_definitions, RelationshipDefinition, Relationship
 from pylms.core import resolve_persons
+from abc import abstractmethod, ABC
+import logging
+
+logger = logging.getLogger(__name__)
 
 
 class ExitPyLMS(BaseException):
  pass
 
 
-def _input_or_exit_pylms():
- try:
- return input()
- except KeyboardInterrupt:
- raise ExitPyLMS()
+class IOs(ABC):
+ @abstractmethod
+ def show_person(self, person: Person) -> None:
+ pass
 
+ @abstractmethod
+ def list_persons(self, persons: list[Person]) -> None:
+ pass
 
-def list_persons() -> None:
- persons = storage.read_persons()
- if persons:
- relationships = storage.read_relationships(persons)
- resolved_persons = resolve_persons(persons, relationships)
+ @abstractmethod
+ def select_person(self, persons: list[Person]) -> Person | None:
+ pass
+
+ @abstractmethod
+ def update_person(self, person_to_update: Person) -> Person:
+ pass
 
- for person, rs in sorted(resolved_persons, key=lambda t: t[0].person_id):
- _print_person(person)
- for r in rs:
- other = r.right if r.left == person else r.left
- print(f" -> {r.repr_for(person)} de ({other.person_id}) {other}")
- else:
- print("No Person registered yet.")
-
-
-def _print_person(person):
- created = person.created
- print(
- f"({person.person_id})",
- person,
- f"({created.year}-{created.month}-{created.day} {created.hour}-{created.minute}-{created.second})",
- )
 
+class EventListener(ABC):
+ @abstractmethod
+ def creating_person(self, person: Person) -> None:
+ pass
 
-def _interactive_person_id(valid_ids: list[int]) -> int:
-if not valid_ids:
-raise ValueError("valid_ids can not be empty.")
+ @abstractmethod
+def deleting_person(self, person_to_delete: Person) -> None:
+pass
 
- while True:
- n = _input_or_exit_pylms()
+ @abstractmethod
+ def creating_link(self, rl_definition: RelationshipDefinition, person_left: Person, person_right: Person) -> None:
+ pass
 
- try:
- res = int(n)
- if res not in valid_ids:
- print("Not a valid id.")
- continue
 
- return res
- except ValueError:
- print("Not an integer.")
+ios: IOs
+events: EventListener
+
+
+def list_persons() -> None:
+ persons = storage.read_persons()
+
+ resolved_persons = []
+ if persons:
+ relationships = storage.read_relationships(persons)
+ resolved_persons = resolve_persons(persons, relationships)
+
+ ios.list_persons(resolved_persons)
 
 
 def _interactive_select_person(pattern: str) -> Person | None:
  persons = _search_person(pattern)
  if not persons:
- print("No match.")
  return
 
  if len(persons) == 1:
  return persons[0]
 
- print("Input id of person to update:")
- for person in sorted(persons, key=lambda p: p.person_id):
- _print_person(person)
- print("CTRL+C to exit")
-
- person_id = _interactive_person_id([person.person_id for person in persons])
- for person in persons:
- if person.person_id == person_id:
- return person
-
- # should not happen
- raise RuntimeError(f"id {person_id} does not exist in list of Persons")
-
-
-def _interactive_person_details() -> tuple[str, str | None]:
- while True:
- text = _input_or_exit_pylms()
-
- words = text.split(" ")
- if len(words) > 2:
- print("Too many words.")
- continue
-
- if len(words) == 1:
- return words[0], None
- return words[0], words[1]
+ return ios.select_person(persons)
 
 
 def update_person(pattern: str) -> None:
  person_to_update: Person = _interactive_select_person(pattern)
  if not person_to_update:
  return
 
- print("Input new first name and last name to update:")
- _print_person(person_to_update)
- print("CTRL+C to exit")
-
- firstname: str
- lastname: str
- firstname, lastname = _interactive_person_details()
+ updated_person = ios.update_person(person_to_update)
 
  persons = storage.read_persons()
  for person in persons:
  if person.person_id == person_to_update.person_id:
- person.firstname = firstname
- person.lastname = lastname
+ person.firstname = updated_person.firstname
+ person.lastname = updated_person.lastname
  storage.store_persons(persons)
 
 
 def search_person(pattern: str) -> None:
  for person in _search_person(pattern):
-_print_person(person)
+ios.show_person(person)
 
 
 def _search_person(pattern: str) -> list[Person]:
@@ -143,39 +113,24 @@ def store_person(firstname: str) -> None:
  persons = storage.read_persons()
  id_generator = PersonIdGenerator(persons)
  person = Person(person_id=id_generator.next_person_id(), firstname=firstname)
-print(f"Create Person {person}.")
+events.creating_person(person)
  storage.store_persons([person] + persons)
 
 
 def store_person(firstname: str, lastname: str = None) -> None:
  persons = storage.read_persons()
  id_generator = PersonIdGenerator(persons)
  person = Person(person_id=id_generator.next_person_id(), firstname=firstname, lastname=lastname)
-print(f"Create Person {person}.")
+events.creating_person(person)
  storage.store_persons([person] + persons)
 
 
-def _interactive_hit_enter():
- while True:
- s = _input_or_exit_pylms()
-
- if len(s) == 0:
- return
-
- print("Just hit ENTER")
- continue
-
-
 def delete_person(pattern: str) -> None:
  person_to_delete: Person = _interactive_select_person(pattern)
  if not person_to_delete:
  return
 
- print("Hit ENTER to delete:")
- _print_person(person_to_delete)
- print("CTRL+C to exit")
-
- _interactive_hit_enter()
+ events.deleting_person(person_to_delete)
 
  persons = storage.read_persons()
  persons.remove(person_to_delete)
@@ -215,7 +170,7 @@ def _parse_nl_link_request(natural_language_link_request: str) -> LinkRequest |
  person_patterns = list(filter(lambda s: len(s) > 0, map(str.strip, natural_language_link_request.split(alias))))
  patterns_count = len(person_patterns)
  if patterns_count != 2:
-print(f"Unsupported link request: wrong number of person patterns ({patterns_count})")
+logger.error(f"Unsupported link request: wrong number of person patterns ({patterns_count})")
  return None
 
  return LinkRequest(
@@ -235,18 +190,13 @@ def link_persons(natural_language_link_request: str) -> None:
  person_right = _interactive_select_person(link_request.right_person_pattern)
 
  if person_left is None:
-print(f'No match for "{link_request.left_person_pattern}".')
+logger.info(f'No match for "{link_request.left_person_pattern}".')
  if person_right is None:
-print(f'No match for "{link_request.right_person_pattern}".')
+logger.info(f'No match for "{link_request.right_person_pattern}".')
  if person_left is None or person_right is None:
  return
 
- print(f'Hit ENTER to link as "{link_request.definition.name}":')
- _print_person(person_left)
- _print_person(person_right)
- print("CTRL+C to exit")
-
- _interactive_hit_enter()
+ events.creating_link(link_request.definition, person_left, person_right)
 
  persons = storage.read_persons()
  relationships = storage.read_relationships(persons)