jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use HassKey in file_upload (#137294)

Browse Source
This commit is contained in:
epenet 2025-02-04 08:45:41 +01:00 committed by GitHub
parent 0f57347797
commit 5e0312ca60
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
homeassistant/components/file_upload/__init__.py
View File

@ -21,9 +21,11 @@ from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import raise_if_invalid_filename
from homeassistant.util.hass_dict import HassKey
from homeassistant.util.ulid import ulid_hex
DOMAIN = "file_upload"
_DATA: HassKey[FileUploadData] = HassKey(DOMAIN)
ONE_MEGABYTE = 1024 * 1024
MAX_SIZE = 100 * ONE_MEGABYTE
@ -41,7 +43,7 @@ def process_uploaded_file(hass: HomeAssistant, file_id: str) -> Iterator[Path]:
if DOMAIN not in hass.data:
raise ValueError("File does not exist")
file_upload_data: FileUploadData = hass.data[DOMAIN]
file_upload_data = hass.data[_DATA]
if not file_upload_data.has_file(file_id):
raise ValueError("File does not exist")
@ -149,10 +151,10 @@ class FileUploadView(HomeAssistantView):
hass = request.app[KEY_HASS]
file_id = ulid_hex()
if DOMAIN not in hass.data:
hass.data[DOMAIN] = await FileUploadData.create(hass)
if _DATA not in hass.data:
hass.data[_DATA] = await FileUploadData.create(hass)
file_upload_data: FileUploadData = hass.data[DOMAIN]
file_upload_data = hass.data[_DATA]
file_dir = file_upload_data.file_dir(file_id)
queue: SimpleQueue[tuple[bytes, asyncio.Future[None] | None] | None] = (
SimpleQueue()
@ -206,7 +208,7 @@ class FileUploadView(HomeAssistantView):
raise web.HTTPNotFound
file_id = data["file_id"]
file_upload_data: FileUploadData = hass.data[DOMAIN]
file_upload_data = hass.data[_DATA]
if file_upload_data.files.pop(file_id, None) is None:
raise web.HTTPNotFound