fix(playground): resolve the 11 ruff violations in backend_manager

playground/backend_manager/app/auth/jwt_handler.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 from fastapi import Depends, HTTPException, Security, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from jose import JWTError, jwt
@@ -23,13 +23,11 @@
 
 def create_access_token(user_id: str, extra_claims: dict | None = None) -> str:
     """Generate a signed JWT access token."""
-    expire = datetime.now(timezone.utc) + timedelta(
-        minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
-    )
+    expire = datetime.now(UTC) + timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
     to_encode = {
         "sub": user_id,
         "exp": expire,
-        "iat": datetime.now(timezone.utc),
+        "iat": datetime.now(UTC),
         "iss": "artemis-backend-manager",
     }
     if extra_claims:

playground/backend_manager/app/auth/otp_service.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 import logging
 import secrets
 from app.config import settings
@@ -38,7 +38,7 @@ def generate_otp(self, identifier: str) -> str:
         else:
             code = f"{secrets.randbelow(900000) + 100000}"
 
-        expires_at = datetime.now(timezone.utc) + timedelta(seconds=settings.OTP_EXPIRE_SECONDS)
+        expires_at = datetime.now(UTC) + timedelta(seconds=settings.OTP_EXPIRE_SECONDS)
         self._store[ident] = {
             "code": code,
             "expires_at": expires_at,
@@ -65,7 +65,7 @@ def verify_otp(self, identifier: str, code: str) -> bool:
             logger.warning(f"[OTP Service] No active OTP found for {ident}")
             return False
 
-        if datetime.now(timezone.utc) > record["expires_at"]:
+        if datetime.now(UTC) > record["expires_at"]:
             logger.warning(f"[OTP Service] OTP for {ident} has expired")
             self._store.pop(ident, None)
             return False

playground/backend_manager/app/services/bigquery_service.py

@@ -14,7 +14,6 @@
 
 from datetime import datetime, timezone
 import logging
-from typing import Optional
 from app.config import settings
 from app.schemas.session_schema import SessionRecord, SessionStatus
 
@@ -34,7 +33,7 @@ class BigQueryMappingService:
 
     def __init__(self):
         self._local_cache: dict[str, SessionRecord] = {}
-        self._client: Optional["bigquery.Client"] = None
+        self._client: bigquery.Client | None = None
         self._table_ref: str = (
             f"{settings.GCP_PROJECT_ID}.{settings.BQ_DATASET}.{settings.BQ_TABLE}"
         )

playground/backend_manager/app/services/docker_service.py

@@ -15,7 +15,6 @@
 import asyncio
 import logging
 import os
-from typing import Optional
 from app.config import settings
 
 logger = logging.getLogger("artemis.docker")
@@ -33,7 +32,7 @@ class DockerManagerService:
     """Manages creation, monitoring, and deletion of ephemeral Artemis session containers via Docker Socket."""
 
     def __init__(self):
-        self._client: Optional["docker.DockerClient"] = None
+        self._client: docker.DockerClient | None = None
         if DOCKER_AVAILABLE:
             try:
                 self._client = docker.DockerClient(base_url=settings.DOCKER_SOCKET_PATH)

playground/backend_manager/app/services/session_manager.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 import asyncio
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 import logging
 import uuid
 from app.config import settings
@@ -49,7 +49,7 @@ async def _reaper_loop(self):
         while True:
             try:
                 await asyncio.sleep(settings.REAPER_CHECK_INTERVAL_SECONDS)
-                now = datetime.now(timezone.utc)
+                now = datetime.now(UTC)
                 active_sessions = await bigquery_service.list_all_active_sessions()
 
                 for s in active_sessions:
@@ -78,7 +78,7 @@ async def _reaper_loop(self):
     async def create_session(self, user_id: str, request: CreateSessionRequest) -> SessionResponse:
         """Provision a new Cuttlefish emulator + Artemis container and link them via ADB."""
         session_id = str(uuid.uuid4())
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
         expires_at = now + timedelta(minutes=request.ttl_minutes)
 
         logger.info(
@@ -164,7 +164,7 @@ async def heartbeat(self, session_id: str, user_id: str) -> HeartbeatResponse |
         if not record or record.user_id != user_id or record.status == SessionStatus.TERMINATED:
             return None
 
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
         record.last_heartbeat_at = now
         record.expires_at = now + timedelta(minutes=settings.SESSION_TTL_MINUTES)
         await bigquery_service.save_session(record)