Files
agent-framework/python/run_tasks_in_packages_if_exists.py
T
Eduard van Valkenburg 3449902b03 Python: added ChatClientBase with function calling (#147)
* added ChatClientBase with function calling

* streaming update

* fixed typing

* test setup

* small update

* src setup

* removed src, updated test naming

* fixed test command

* alolow args

* updated test run

* added unit test folder to azure

* added init and unit test to azure

* added other cross tests

* restructured

* reset test run

* fix name

* removed always

* updated test

* extend pytest.xml locations

* run surface always

* added decorators for FC and marked tests

* fixed mypy settings and added tests

* fix override import

* removed import
2025-07-10 09:18:15 +00:00

78 lines
2.3 KiB
Python

# Copyright (c) Microsoft. All rights reserved.
import glob
import sys
from pathlib import Path
import tomli
from poethepoet.app import PoeThePoet
from rich import print
def discover_projects(workspace_pyproject_file: Path) -> list[Path]:
with workspace_pyproject_file.open("rb") as f:
data = tomli.load(f)
projects = data["tool"]["uv"]["workspace"]["members"]
exclude = data["tool"]["uv"]["workspace"].get("exclude", [])
all_projects: list[Path] = []
for project in projects:
if "*" in project:
globbed = glob.glob(str(project), root_dir=workspace_pyproject_file.parent)
globbed_paths = [Path(p) for p in globbed]
all_projects.extend(globbed_paths)
else:
all_projects.append(Path(project))
for project in exclude:
if "*" in project:
globbed = glob.glob(str(project), root_dir=workspace_pyproject_file.parent)
globbed_paths = [Path(p) for p in globbed]
all_projects = [p for p in all_projects if p not in globbed_paths]
else:
all_projects = [p for p in all_projects if p != Path(project)]
return all_projects
def extract_poe_tasks(file: Path) -> set[str]:
with file.open("rb") as f:
data = tomli.load(f)
tasks = set(data.get("tool", {}).get("poe", {}).get("tasks", {}).keys())
# Check if there is an include too
include: str | None = data.get("tool", {}).get("poe", {}).get("include", None)
if include:
include_file = file.parent / include
if include_file.exists():
tasks = tasks.union(extract_poe_tasks(include_file))
return tasks
def main() -> None:
pyproject_file = Path(__file__).parent / "pyproject.toml"
projects = discover_projects(pyproject_file)
if len(sys.argv) < 2:
print("Please provide a task name")
sys.exit(1)
task_name = sys.argv[1]
for project in projects:
tasks = extract_poe_tasks(project / "pyproject.toml")
if task_name in tasks:
print(f"Running task {task_name} in {project}")
app = PoeThePoet(cwd=project)
result = app(cli_args=sys.argv[1:])
if result:
sys.exit(result)
else:
print(f"Task {task_name} not found in {project}")
if __name__ == "__main__":
main()