44quick switching between different PR versions without rebuilding.
55"""
66
7+ from __future__ import annotations
8+
79import json
810import shutil
911from datetime import datetime , timedelta
1012from pathlib import Path
11- from typing import Dict , List , Optional # noqa: UP035
1213
1314from rich import print as rprint
1415
@@ -95,7 +96,7 @@ def save_cache_info(self, pr_info, cache_path: Path) -> None:
9596 # Enforce cache limits after saving new cache
9697 self .enforce_cache_limits ()
9798
98- def get_cached_frontend_path (self , pr_info ) -> Optional [ Path ] :
99+ def get_cached_frontend_path (self , pr_info ) -> Path | None :
99100 """Get path to cached frontend build if valid"""
100101 cache_path = self .get_frontend_cache_path (pr_info )
101102 dist_path = cache_path / "repo" / "dist"
@@ -104,7 +105,7 @@ def get_cached_frontend_path(self, pr_info) -> Optional[Path]:
104105 return dist_path
105106 return None
106107
107- def _load_cache_info (self , cache_dir : Path ) -> Optional [ dict ] :
108+ def _load_cache_info (self , cache_dir : Path ) -> dict | None :
108109 """Load cache info from a directory."""
109110 info_path = self .get_cache_info_path (cache_dir )
110111 if not info_path .exists ():
@@ -127,7 +128,7 @@ def _clean_specific_pr_cache(self, frontend_cache: Path, pr_number: int) -> None
127128 shutil .rmtree (cache_dir )
128129 break
129130
130- def clean_frontend_cache (self , pr_number : Optional [ int ] = None ) -> None :
131+ def clean_frontend_cache (self , pr_number : int | None = None ) -> None :
131132 """Clean frontend cache (specific PR or all)."""
132133 frontend_cache = self .cache_dir / "frontend"
133134 if not frontend_cache .exists ():
@@ -145,15 +146,15 @@ def _calculate_cache_size_mb(self, cache_dir: Path) -> float:
145146 total_size = sum (f .stat ().st_size for f in cache_dir .rglob ("*" ) if f .is_file ())
146147 return total_size / (1024 * 1024 )
147148
148- def _get_cache_info_with_metadata (self , cache_dir : Path ) -> Optional [ dict ] :
149+ def _get_cache_info_with_metadata (self , cache_dir : Path ) -> dict | None :
149150 """Get cache info with additional metadata like path and size."""
150151 info = self ._load_cache_info (cache_dir )
151152 if info :
152153 info ["cache_path" ] = str (cache_dir )
153154 info ["size_mb" ] = self ._calculate_cache_size_mb (cache_dir )
154155 return info
155156
156- def list_cached_frontends (self ) -> List [ Dict ]: # noqa: UP006
157+ def list_cached_frontends (self ) -> list [ dict ]:
157158 """List all cached frontend PRs."""
158159 frontend_cache = self .cache_dir / "frontend"
159160 if not frontend_cache .exists ():
@@ -177,7 +178,7 @@ def _is_cache_expired(self, cached_at: str) -> bool:
177178 except (ValueError , TypeError ):
178179 return True # Consider invalid timestamps as expired
179180
180- def _get_expired_items (self , cached_items : List [ Dict ]) -> List [ Dict ]: # noqa: UP006
181+ def _get_expired_items (self , cached_items : list [ dict ]) -> list [ dict ]:
181182 """Get list of expired cache items."""
182183 expired = []
183184 for item in cached_items :
@@ -186,7 +187,7 @@ def _get_expired_items(self, cached_items: List[Dict]) -> List[Dict]: # noqa: U
186187 expired .append (item )
187188 return expired
188189
189- def _get_excess_items (self , cached_items : List [ Dict ], expired_items : List [ Dict ]) -> List [ Dict ]: # noqa: UP006
190+ def _get_excess_items (self , cached_items : list [ dict ], expired_items : list [ dict ]) -> list [ dict ]:
190191 """Get list of items that exceed the maximum cache limit."""
191192 remaining_items = [item for item in cached_items if item not in expired_items ]
192193 if len (remaining_items ) > self .max_cache_items :
0 commit comments