get binary option from http api.
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form
|
||||
from fastapi.responses import PlainTextResponse, Response
|
||||
from fastapi.responses import PlainTextResponse, Response, JSONResponse
|
||||
from typing import List, Optional
|
||||
from datetime import datetime
|
||||
from uuid import UUID
|
||||
@@ -423,3 +423,67 @@ async def get_object_text(
|
||||
raise HTTPException(status_code=500, detail="Failed to retrieve text object")
|
||||
|
||||
return PlainTextResponse(content=content, media_type="text/plain; charset=utf-8")
|
||||
|
||||
class ObjectBinaryResponse(BaseModel):
|
||||
uuid: UUID
|
||||
name: str
|
||||
binary: bool
|
||||
size: int
|
||||
content_type: str
|
||||
data_base64: str
|
||||
private: bool
|
||||
created_at: datetime
|
||||
modified_at: datetime
|
||||
|
||||
@router.get("/objects/{uuid}/binary", response_model=ObjectBinaryResponse)
|
||||
async def get_object_binary(
|
||||
uuid: UUID,
|
||||
db: DbDependency,
|
||||
current_user: HttpUser = Depends(get_current_http_user)
|
||||
):
|
||||
username = current_user.username
|
||||
|
||||
try:
|
||||
with db.transaction() as conn:
|
||||
root = conn.root()
|
||||
|
||||
obj = Object.get_object_by_uuid(uuid, root)
|
||||
if not obj:
|
||||
raise HTTPException(status_code=404, detail="Object not found")
|
||||
|
||||
# Authorization check for private objects
|
||||
if obj.private:
|
||||
user = User.get_user_by_username(username, root)
|
||||
if not user or user.uuid != obj.owner:
|
||||
raise HTTPException(status_code=403, detail="Not authorized to access this private object")
|
||||
|
||||
# Get content as bytes (works for both text and binary)
|
||||
content_bytes = obj.data_bytes # uses the property that always returns bytes
|
||||
|
||||
# Encode to base64
|
||||
data_base64 = base64.b64encode(content_bytes).decode('ascii')
|
||||
|
||||
# Guess content_type
|
||||
content_type, _ = mimetypes.guess_type(obj.name)
|
||||
if content_type is None:
|
||||
content_type = "application/octet-stream" if obj.binary else "text/plain"
|
||||
|
||||
logging.info(f"User {username} downloaded binary/base64 object {uuid} ({obj.name})")
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logging.error(f"Binary download failed for {username} on {uuid}: {e}\n{traceback.format_exc()}")
|
||||
raise HTTPException(status_code=500, detail="Failed to retrieve object")
|
||||
|
||||
return ObjectBinaryResponse(
|
||||
uuid=obj.uuid,
|
||||
name=obj.name,
|
||||
binary=obj.binary,
|
||||
size=obj.size,
|
||||
content_type=content_type,
|
||||
data_base64=data_base64,
|
||||
private=obj.private,
|
||||
created_at=obj.created_at,
|
||||
modified_at=obj.modified_at
|
||||
)
|
||||
@@ -78,6 +78,10 @@ class Object(persistent.Persistent):
|
||||
self._binary = False
|
||||
self.touch()
|
||||
|
||||
@property
|
||||
def data_bytes(self):
|
||||
return self._data
|
||||
|
||||
@property
|
||||
def owner(self) -> Optional[UUID]:
|
||||
return self._owner
|
||||
|
||||
Reference in New Issue
Block a user