Skip to content

Refactor app.py: improve type annotations and extract helper functions #177

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

Merged
redreceipt merged 3 commits into from
Dec 24, 2025
Merged

Refactor app.py: improve type annotations and extract helper functions #177

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7dd4c62
Initial plan
Copilot Dec 24, 2025
96a5fb0
Apply refactoring diffs: TypeVar rename, IndexFutures, constants, and…
Copilot Dec 24, 2025
0957588
Add type parameters to Future annotations for better type safety
Copilot Dec 24, 2025
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
147 changes: 87 additions & 60 deletions app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,16 @@ def normalize_identity(value: str | None) -> str:
return re.sub(r"[^a-z0-9]", "", value.lower())


def format_display_name(linear_username: str) -> str:
return re.sub(r"[._-]+", " ", linear_username).title()


app = Flask(__name__)

# Maximum number of distinct _build_index_context results to cache.
# This can be increased or made configurable based on production usage patterns.
INDEX_CONTEXT_CACHE_MAXSIZE = 16


def record_breakdown(
store_points: dict[str, dict[str, int]],
Expand Down Expand Up @@ -150,14 +158,29 @@ def mmdd_filter(date_str: str) -> str:
return date_str


T = TypeVar('T')
ResultType = TypeVar('ResultType')


class IndexFutures(TypedDict):
"""Named collection of futures used for building the index view."""
created_priority: Future[list]
open_priority: Future[list]
completed_priority: Future[list]
completed_bugs: Future[list]
completed_new_features: Future[list]
completed_technical_changes: Future[list]
open_bugs: Future[list]
open_new_features: Future[list]
open_technical_changes: Future[list]
merged_prs_by_reviewer: Future[dict]
merged_prs_by_author: Future[dict]


def get_future_result_with_timeout(
future: Future[T],
default_value: T,
future: Future[ResultType],
default_value: ResultType,
timeout: int = INDEX_FUTURE_TIMEOUT
) -> T:
) -> ResultType:
"""
Get result from a future with a timeout, returning a default value on timeout.

Expand All @@ -178,19 +201,7 @@ def get_future_result_with_timeout(
def _submit_index_futures(
executor: ThreadPoolExecutor,
days: int,
) -> tuple[
Future,
Future,
Future,
Future,
Future,
Future,
Future,
Future,
Future,
Future,
Future,
]:
) -> IndexFutures:
"""
Submit all futures required to build the index context.

Expand Down Expand Up @@ -223,40 +234,37 @@ def _submit_index_futures(
reviews_future = executor.submit(merged_prs_by_reviewer, days)
authored_prs_future = executor.submit(merged_prs_by_author, days)

return (
created_priority_future,
open_priority_future,
completed_priority_future,
completed_bugs_future,
completed_new_features_future,
completed_technical_changes_future,
open_bugs_future,
open_new_features_future,
open_technical_changes_future,
reviews_future,
authored_prs_future,
)
return {
"created_priority": created_priority_future,
"open_priority": open_priority_future,
"completed_priority": completed_priority_future,
"completed_bugs": completed_bugs_future,
"completed_new_features": completed_new_features_future,
"completed_technical_changes": completed_technical_changes_future,
"open_bugs": open_bugs_future,
"open_new_features": open_new_features_future,
"open_technical_changes": open_technical_changes_future,
"merged_prs_by_reviewer": reviews_future,
"merged_prs_by_author": authored_prs_future,
}


@lru_cache(maxsize=16)
def _build_index_context(days: int, _cache_epoch: int) -> dict:
with ThreadPoolExecutor(max_workers=INDEX_THREADPOOL_MAX_WORKERS) as executor:
(
created_priority_future,
open_priority_future,
completed_priority_future,
completed_bugs_future,
completed_new_features_future,
completed_technical_changes_future,
open_bugs_future,
open_new_features_future,
open_technical_changes_future,
reviews_future,
authored_prs_future,
) = _submit_index_futures(executor, days)

created_priority_bugs = get_future_result_with_timeout(created_priority_future, [])
open_priority_bugs = get_future_result_with_timeout(open_priority_future, [])
def _get_priority_bugs_from_futures(
created_priority_future: Future[list],
open_priority_future: Future[list],
completed_priority_future: Future[list],
) -> tuple[list, list, list]:
"""
Retrieve priority bug data from futures and filter completed issues.

Only non-project issues are included in the completed list.
"""
created_priority_bugs = get_future_result_with_timeout(
created_priority_future, []
)
open_priority_bugs = get_future_result_with_timeout(
open_priority_future, []
)

# Only include non-project issues in the index summary
completed_priority_result = get_future_result_with_timeout(
Expand All @@ -267,32 +275,54 @@ def _build_index_context(days: int, _cache_epoch: int) -> dict:
for issue in completed_priority_result
if not issue.get("project")
]
completed_bugs_result = get_future_result_with_timeout(completed_bugs_future, [])

return created_priority_bugs, open_priority_bugs, completed_priority_bugs


@lru_cache(maxsize=INDEX_CONTEXT_CACHE_MAXSIZE)
def _build_index_context(days: int, _cache_epoch: int) -> dict:
with ThreadPoolExecutor(max_workers=INDEX_THREADPOOL_MAX_WORKERS) as executor:
futures = _submit_index_futures(executor, days)

(
created_priority_bugs,
open_priority_bugs,
completed_priority_bugs,
) = _get_priority_bugs_from_futures(
futures["created_priority"],
futures["open_priority"],
futures["completed_priority"],
)
completed_bugs_result = get_future_result_with_timeout(
futures["completed_bugs"], []
)
completed_bugs = [
issue
for issue in completed_bugs_result
if not issue.get("project")
]
completed_new_features_result = get_future_result_with_timeout(
completed_new_features_future, []
futures["completed_new_features"], []
)
completed_new_features = [
issue
for issue in completed_new_features_result
if not issue.get("project")
]
completed_technical_changes_result = get_future_result_with_timeout(
completed_technical_changes_future, []
futures["completed_technical_changes"], []
)
completed_technical_changes = [
issue
for issue in completed_technical_changes_result
if not issue.get("project")
]
open_bugs_result = get_future_result_with_timeout(open_bugs_future, [])
open_new_features_result = get_future_result_with_timeout(open_new_features_future, [])
open_bugs_result = get_future_result_with_timeout(futures["open_bugs"], [])
open_new_features_result = get_future_result_with_timeout(
futures["open_new_features"], []
)
open_technical_changes_result = get_future_result_with_timeout(
open_technical_changes_future, []
futures["open_technical_changes"], []
)
open_work = (
open_bugs_result
Expand All @@ -313,9 +343,6 @@ def _build_index_context(days: int, _cache_epoch: int) -> dict:
if info.get("team") == "apollos_engineering"
}

def format_display_name(linear_username: str) -> str:
return re.sub(r"[._-]+", " ", linear_username).title()

alias_to_slug = {}
github_to_slug = {}
display_name_overrides = {}
Expand Down Expand Up @@ -398,8 +425,8 @@ def resolve_slug(*identities: str | None) -> str | None:
1,
)

merged_reviews = get_future_result_with_timeout(reviews_future, {})
merged_authored_prs = get_future_result_with_timeout(authored_prs_future, {})
merged_reviews = get_future_result_with_timeout(futures["merged_prs_by_reviewer"], {})
merged_authored_prs = get_future_result_with_timeout(futures["merged_prs_by_author"], {})

for reviewer, prs in merged_reviews.items():
review_points = len(prs)
Expand Down