modify process_groups_list so it can process any group path, not just the root
process_groups_list now takes an optional group path
This commit is contained in:
parent
677ffc4130
commit
df43eb0b82
|
@ -141,6 +141,12 @@ paths:
|
|||
|
||||
/process-groups:
|
||||
parameters:
|
||||
- name: process_group_identifier
|
||||
in: query
|
||||
required: false
|
||||
description: Optional parameter to filter by a single group
|
||||
schema:
|
||||
type: string
|
||||
- name: page
|
||||
in: query
|
||||
required: false
|
||||
|
|
|
@ -178,9 +178,12 @@ def process_group_update(
|
|||
return make_response(jsonify(process_group), 200)
|
||||
|
||||
|
||||
def process_groups_list(page: int = 1, per_page: int = 100) -> flask.wrappers.Response:
|
||||
def process_groups_list(process_group_identifier: str = None, page: int = 1, per_page: int = 100) -> flask.wrappers.Response:
|
||||
"""Process_groups_list."""
|
||||
process_groups = ProcessModelService().get_process_groups()
|
||||
if process_group_identifier is not None:
|
||||
process_groups = ProcessModelService().get_process_groups(process_group_identifier)
|
||||
else:
|
||||
process_groups = ProcessModelService().get_process_groups()
|
||||
batch = ProcessModelService().get_batch(
|
||||
items=process_groups, page=page, per_page=per_page
|
||||
)
|
||||
|
|
|
@ -160,9 +160,9 @@ class ProcessModelService(FileSystemService):
|
|||
process_models.sort()
|
||||
return process_models
|
||||
|
||||
def get_process_groups(self) -> list[ProcessGroup]:
|
||||
def get_process_groups(self, process_group_id: str = None) -> list[ProcessGroup]:
|
||||
"""Returns the process_groups as a list in display order."""
|
||||
process_groups = self.__scan_process_groups()
|
||||
process_groups = self.__scan_process_groups(process_group_id)
|
||||
process_groups.sort()
|
||||
return process_groups
|
||||
|
||||
|
@ -254,12 +254,16 @@ class ProcessModelService(FileSystemService):
|
|||
index += 1
|
||||
return process_groups
|
||||
|
||||
def __scan_process_groups(self) -> list[ProcessGroup]:
|
||||
def __scan_process_groups(self, process_group_id: str = None) -> list[ProcessGroup]:
|
||||
"""__scan_process_groups."""
|
||||
if not os.path.exists(FileSystemService.root_path()):
|
||||
return [] # Nothing to scan yet. There are no files.
|
||||
if process_group_id is not None:
|
||||
scan_path = os.path.join(FileSystemService.root_path(), process_group_id)
|
||||
else:
|
||||
scan_path = FileSystemService.root_path()
|
||||
|
||||
with os.scandir(FileSystemService.root_path()) as directory_items:
|
||||
with os.scandir(scan_path) as directory_items:
|
||||
process_groups = []
|
||||
for item in directory_items:
|
||||
# if item.is_dir() and not item.name[0] == ".":
|
||||
|
|
Loading…
Reference in New Issue