chore(blocks/fal): Use dict instead of Dict (#8855)

Replace Dict with dict

### Changes 🏗️

Replace Dict with dict

### Checklist 📋

#### For code changes:
- [ ] I have clearly listed my changes in the PR description
- [ ] I have made a test plan
- [ ] I have tested my changes according to the test plan:
  <!-- Put your test plan here: -->
  - [ ] ...

<details>
  <summary>Example test plan</summary>
  
  - [ ] Create from scratch and execute an agent with at least 3 blocks
- [ ] Import an agent from file upload, and confirm it executes
correctly
  - [ ] Upload agent to marketplace
- [ ] Import an agent from marketplace and confirm it executes correctly
  - [ ] Edit an agent from monitor, and confirm it executes correctly
</details>

#### For configuration changes:
- [ ] `.env.example` is updated or already compatible with my changes
- [ ] `docker-compose.yml` is updated or already compatible with my
changes
- [ ] I have included a list of my configuration changes in the PR
description (under **Changes**)

<details>
  <summary>Examples of configuration changes</summary>

  - Changing ports
  - Adding new services that need to communicate with each other
  - Secrets or environment variable changes
  - New or infrastructure changes such as databases
</details>
This commit is contained in:
Aarushi 2024-12-03 11:09:38 +00:00 committed by GitHub
parent 3bca279b35
commit de1cd6c295
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 6 additions and 6 deletions

View File

@ -16,7 +16,7 @@ class RateLimiter:
):
self.redis = Redis(
host=redis_host,
port=redis_port,
port=int(redis_port),
password=redis_password,
decode_responses=True,
)

View File

@ -1,7 +1,7 @@
import logging
import time
from enum import Enum
from typing import Any, Dict
from typing import Any
import httpx
@ -64,7 +64,7 @@ class AIVideoGeneratorBlock(Block):
},
)
def _get_headers(self, api_key: str) -> Dict[str, str]:
def _get_headers(self, api_key: str) -> dict[str, str]:
"""Get headers for FAL API requests."""
return {
"Authorization": f"Key {api_key}",
@ -72,8 +72,8 @@ class AIVideoGeneratorBlock(Block):
}
def _submit_request(
self, url: str, headers: Dict[str, str], data: Dict[str, Any]
) -> Dict[str, Any]:
self, url: str, headers: dict[str, str], data: dict[str, Any]
) -> dict[str, Any]:
"""Submit a request to the FAL API."""
try:
response = httpx.post(url, headers=headers, json=data)
@ -83,7 +83,7 @@ class AIVideoGeneratorBlock(Block):
logger.error(f"FAL API request failed: {str(e)}")
raise RuntimeError(f"Failed to submit request: {str(e)}")
def _poll_status(self, status_url: str, headers: Dict[str, str]) -> Dict[str, Any]:
def _poll_status(self, status_url: str, headers: dict[str, str]) -> dict[str, Any]:
"""Poll the status endpoint until completion or failure."""
try:
response = httpx.get(status_url, headers=headers)