Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[24.1] Add TypedDict for JobsSummary #18418

Merged
merged 1 commit into from
Jun 18, 2024
Merged

[24.1] Add TypedDict for JobsSummary #18418

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions lib/galaxy/managers/jobs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
cast,
Dict,
List,
Optional,
)

import sqlalchemy
Expand All @@ -26,6 +27,7 @@
)
from sqlalchemy.orm import aliased
from sqlalchemy.sql import select
from typing_extensions import TypedDict
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It can be imported from typing since 24.1 is Python>=3.8 and this is not required by Pulsar?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are multiple other locations where we're importing TypedDict from typing_extensions, I'll make a pass at them in a separate PR.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks!


from galaxy import model
from galaxy.exceptions import (
Expand Down Expand Up @@ -824,7 +826,14 @@ def merge_states(component_states):
return rval


def summarize_jobs_to_dict(sa_session, jobs_source):
class JobsSummary(TypedDict):
populated_state: str
states: Dict[str, int]
model: str
id: int


def summarize_jobs_to_dict(sa_session, jobs_source) -> Optional[JobsSummary]:
"""Produce a summary of jobs for job summary endpoints.

:type jobs_source: a Job or ImplicitCollectionJobs or None
Expand All @@ -833,7 +842,7 @@ def summarize_jobs_to_dict(sa_session, jobs_source):
:rtype: dict
:returns: dictionary containing job summary information
"""
rval = None
rval: Optional[JobsSummary] = None
if jobs_source is None:
pass
elif isinstance(jobs_source, model.Job):
Expand Down
Loading