academon/core/classes/replication.py

147 lines
5.4 KiB
Python

import sqlite3
import pathlib
import uuid
from typing import Union, Optional
from contentmonster.classes.file import File as ContentMonsterFile
from contentmonster.classes.directory import Directory as ContentMonsterDirectory
from contentmonster.classes.vessel import Vessel as ContentMonsterVessel
from ..models.replication import ReplicationFile, ReplicationSource, ReplicationFileLog, ReplicationTarget
class ContentMonsterDatabase:
"""Class wrapping Django database for ContentMonster
"""
def commit(self) -> None:
"""noop
"""
pass
def getFileUUID(self, fileobj: ContentMonsterFile) -> str:
"""Retrieve unique identifier for ContentMonsterFile object
Args:
fileobj (ContentMonsterFile): ContentMonsterFile object to retrieve UUID for
Returns:
str: UUID for passed ContentMonsterFile object
"""
hash = fileobj.getHash()
files = ReplicationFile.objects.filter(directory__name=fileobj.directory.name, name=fileobj.name)
# If file with same name and directory exists
for result in files:
# If it has the same hash, it is the same file -> return its UUID
if file.checksum == hash:
fileuuid = result.uuid
# If not, it is a file that can no longer exist -> delete it
else:
self.removeFileByUUID(result.uuid)
# Return found UUID or generate a new one
return fileuuid or self.addFile(fileobj, hash)
def addFile(self, fileobj: ContentMonsterFile, hash: Optional[str] = None) -> str:
"""Adds a new ReplicationFile object to the database
Args:
fileobj (ContentMonsterFile): ContentMonsterFile object to add to database
hash (str, optional): Checksum of the file, if already known.
Defaults to None and will use .getHash() to calculate checksum then.
Returns:
str: UUID of the new ContentMonsterFile record
"""
hash = hash or fileobj.getHash()
fileuuid = str(uuid.uuid4())
directory = ReplicationSource.objects.get(name=fileobj.directory.name)
ReplicationFile.objects.create(uuid=fileuuid, directory=directory, name=fileobj.name, checksum=hash)
return fileuuid
def getFileByUUID(self, fileuuid: str) -> Optional[tuple[str, str, str]]:
"""Get additional information on a ContentMonsterFile by its UUID
Args:
fileuuid (str): The UUID of the ReplicationFile to retrieve from the database
Returns:
tuple: A tuple consisting of (directory, name, checksum), where
"directory" is the name of the Directory object the File is
located in, "name" is the filename (basename) of the File and
checksum is the SHA256 hash of the file at the time of insertion
into the database. None is returned if no such record is found.
"""
try:
result = ReplicationFile.objects.get(uuid=fileuuid)
return (result.directory.name, result.name, result.checksum)
except ReplicationFile.DoesNotExist:
return None
def removeFile(self, directory: ContentMonsterDirectory, name: str) -> None:
"""Remove a ReplicationFile from the database based on ContentMonsterDirectory and filename
Args:
directory (ContentMonsterDirectory): ContentMonsterDirectory object
containing the ContentMonsterFile to remove
name (str): Filename of the ContentMonsterFile to remove
"""
ReplicationFile.objects.filter(directory__name=directory.name, name=name).delete()
def removeFileByUUID(self, fileuuid: str) -> None:
"""Remove a ReplicationFile from the database based on UUID
Args:
fileuuid (str): The UUID of the ContentMonsterFile to remove from the database
"""
ReplicationFile.objects.filter(uuid=fileuuid).delete()
def logCompletion(self, file: ContentMonsterFile, vessel: ContentMonsterVessel):
"""Log the completion of a ContentMonsterFile upload
Args:
file (ContentMonsterFile): The ContentMonsterFile object that has been uploaded
vessel (ContentMonsterVessel): The ContentMonsterVessel the File has been
uploaded to
"""
fileobj = ReplicationFile.objects.get(uuid=file.uuid)
vesselobj = ReplicationTarget.objects.get(name=vessel.name)
ReplicationFileLog.objects.create(file=fileobj, vessel=vesselobj)
def getCompletionForVessel(self, vessel: ContentMonsterVessel) -> list[Optional[str]]:
"""Get completed uploads for a vessel
Args:
vessel (ContentMonsterVessel): The ContentMonsterVessel object to retrieve
uploaded files for
Returns:
list: List of UUIDs of ContentMonsterFiles that have been successfully uploaded
"""
vesselobj = ReplicationTarget.objects.get(name=vessel.name)
objects = ReplicationFileLog.objects.filter(vessel=vesselobj)
return [o.file.uuid for o in objects]
def getCompletionByFileUUID(self, fileuuid: str) -> list[Optional[str]]:
objects = ReplicationFileLog.objects.filter(file__uuid=fileuuid)
return [o.vessel.name for o in objects]
def migrate(self) -> None:
"""noop
"""
pass
def __del__(self):
"""noop
"""
pass