sio-2425/delivery1/server/services/files.py

59 lines
1.9 KiB
Python

import os
from datetime import datetime
from typing import List, Type
from database import db
from models import File, Organization, User
class FileService:
@staticmethod
def create_dummy_file(org: Organization, user: User) -> File:
file = File(
file_handle = "dummy_file",
document_handle = "org/dummy_file.txt",
name = "dummy_file",
created_at = int(datetime.now().timestamp()),
org_id = 1,
creator_id = 1,
org = org,
creator = user
)
file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "repository"), file.document_handle)
with open(file_path, "w") as f:
f.write("Dummy file content")
db.add(file)
db.commit()
db.refresh(file)
return file
@staticmethod
def get_file(file_id: int) -> File | None:
return db.query(File).filter(File.id == file_id).first()
@staticmethod
def get_file_by_document_handle(document_handle: str) -> File | None:
return db.query(File).filter(File.document_handle == document_handle).first()
@staticmethod
def get_file_by_file_handle(file_handle: str) -> File | None:
return db.query(File).filter(File.file_handle == file_handle).first()
@staticmethod
def list_files() -> list[Type[File]]:
return db.query(File).all()
@staticmethod
def list_files_in_org(org: Organization) -> list[Type[File]]:
return db.query(File).filter(File.org_id == org.id).all()
@staticmethod
def delete_file(file: File) -> File:
file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "repository"), file.document_handle)
os.remove(file_path)
file.document_handle = None
db.commit()
db.refresh(file)
return file