mirror of
https://github.com/flibusta-apps/telegram_files_server.git
synced 2025-12-06 04:25:38 +01:00
96 lines
2.4 KiB
Python
96 lines
2.4 KiB
Python
from typing import Optional
|
|
|
|
from fastapi import (
|
|
File,
|
|
UploadFile,
|
|
Depends,
|
|
Form,
|
|
APIRouter,
|
|
HTTPException,
|
|
Response,
|
|
status,
|
|
)
|
|
|
|
from app.depends import check_token
|
|
from app.models import UploadedFile as UploadedFileDB
|
|
from app.serializers import UploadedFile, CreateUploadedFile
|
|
from app.services.file_downloader import FileDownloader
|
|
from app.services.file_uploader import FileUploader
|
|
|
|
|
|
router = APIRouter(
|
|
prefix="/api/v1/files", dependencies=[Depends(check_token)], tags=["files"]
|
|
)
|
|
|
|
|
|
@router.get("/", response_model=list[UploadedFile])
|
|
async def get_files():
|
|
return await UploadedFileDB.objects.all()
|
|
|
|
|
|
@router.get(
|
|
"/{file_id}",
|
|
response_model=UploadedFile,
|
|
responses={
|
|
404: {},
|
|
},
|
|
)
|
|
async def get_file(file_id: int):
|
|
uploaded_file = await UploadedFileDB.objects.get_or_none(id=file_id)
|
|
|
|
if not uploaded_file:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND)
|
|
|
|
return uploaded_file
|
|
|
|
|
|
@router.post("/", response_model=UploadedFile)
|
|
async def create_file(data: CreateUploadedFile):
|
|
return await UploadedFileDB.objects.create(**data.dict())
|
|
|
|
|
|
@router.post("/upload/", response_model=UploadedFile)
|
|
async def upload_file(file: UploadFile = File({}), caption: Optional[str] = Form({})):
|
|
return await FileUploader.upload(file, caption=caption)
|
|
|
|
|
|
@router.get("/download_by_file_id/{file_id}")
|
|
async def download_by_file_id(file_id: str):
|
|
data = await FileDownloader.download_by_file_id(file_id)
|
|
|
|
if data is None:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST)
|
|
|
|
return Response(data.read())
|
|
|
|
|
|
@router.get("/download_by_message/{chat_id}/{message_id}")
|
|
async def download_by_message(chat_id: str, message_id: int):
|
|
data = await FileDownloader.download_by_message_id(message_id)
|
|
|
|
if data is None:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST)
|
|
|
|
return Response(data.read())
|
|
|
|
|
|
@router.delete("/{file_id}", response_model=UploadedFile, responses={400: {}})
|
|
async def delete_file(file_id: int):
|
|
uploaded_file = await UploadedFileDB.objects.get_or_none(id=file_id)
|
|
|
|
if not uploaded_file:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST)
|
|
|
|
await uploaded_file.delete()
|
|
return uploaded_file
|
|
|
|
|
|
healthcheck_router = APIRouter(
|
|
prefix="/api/v1", dependencies=[Depends(check_token)], tags=["healthcheck"]
|
|
)
|
|
|
|
|
|
@healthcheck_router.get("/healthcheck")
|
|
async def healthcheck():
|
|
return "Ok"
|