import os from datetime import datetime from typing import List, Type from sqlalchemy.orm import Session from models import File, Organization, User class FileService: def __init__(self, db: Session): self.db = db self.repos = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "repository") def create_dummy_file(self, 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(self.repos, file.document_handle) with open(file_path, "w") as f: f.write("Dummy file content") self.db.add(file) self.db.commit() self.db.refresh(file) return file def get_file(self, file_id: int) -> File | None: return self.db.query(File).filter(File.id == file_id).first() def get_file_by_document_handle(self, document_handle: str) -> File | None: return self.db.query(File).filter(File.document_handle == document_handle).first() def get_file_by_file_handle(self, file_handle: str) -> File | None: return self.db.query(File).filter(File.file_handle == file_handle).first() def list_files(self) -> list[Type[File]]: return self.db.query(File).all() def list_files_in_org(self, org: Organization) -> list[Type[File]]: return self.db.query(File).filter(File.org_id == org.id).all() def delete_file(self, file: File) -> File: file_path = os.path.join(self.repos, file.document_handle) os.remove(file_path) file.document_handle = None self.db.commit() self.db.refresh(file) return file