src/materia/models/__init__.py

@@ -16,8 +16,6 @@ from materia.models.database import (
 
 from materia.models.user import User, UserCredentials, UserInfo
 
-from materia.models.filesystem import FileSystem
-
 from materia.models.repository import (
     Repository,
     RepositoryInfo,

src/materia/models/base.py

@@ -1,27 +1,3 @@
-from typing import Optional, Self
+from sqlalchemy.orm import declarative_base
-from sqlalchemy.orm import DeclarativeBase
 
-
+Base = declarative_base()
-class Base(DeclarativeBase):
-
-    def to_dict(self) -> dict:
-        return {key: getattr(self, key) for key in self.__table__.columns.keys()}
-
-    def clone(self) -> Optional[Self]:
-        """Clone model.
-        Included: columns and values, foreign keys
-        Ignored: primary keys, relationships
-        """
-        # if not inspect(self).persistent:
-        #    return
-
-        cloned = self.__class__(
-            **{
-                key: getattr(self, key)
-                for key in self.__table__.columns.keys()
-                # ignore primary keys
-                if key not in self.__table__.primary_key.columns.keys()
-            }
-        )
-
-        return cloned

src/materia/models/directory.py

@@ -47,19 +47,22 @@ class Directory(Base):
         await session.flush()
         await session.refresh(self, attribute_names=["repository"])
 
-        repository_path = await self.repository.path(session, config)
+        relative_path = await self.relative_path(session)
         directory_path = await self.path(session, config)
 
-        new_directory = FileSystem(directory_path, repository_path)
+        try:
-        await new_directory.make_directory()
+            directory_path.mkdir()
+        except OSError as e:
+            raise DirectoryError(
+                f"Failed to create directory at /{relative_path}:",
+                *e.args,
+            )
 
         return self
 
     async def remove(self, session: SessionContext, config: Config):
         session.add(self)
-        await session.refresh(
+        await session.refresh(self, attribute_names=["directories", "files"])
-            self, attribute_names=["repository", "directories", "files"]
-        )
 
         if self.directories:
             for directory in self.directories:
@@ -69,11 +72,15 @@ class Directory(Base):
             for file in self.files:
                 file.remove(session, config)
 
-        repository_path = await self.repository.path(session, config)
+        relative_path = await self.relative_path(session)
         directory_path = await self.path(session, config)
 
-        current_directory = FileSystem(directory_path, repository_path)
+        try:
-        await current_directory.remove()
+            shutil.rmtree(str(directory_path))
+        except OSError as e:
+            raise DirectoryError(
+                f"Failed to remove directory at /{relative_path}:", *e.args
+            )
 
         await session.delete(self)
         await session.flush()
@@ -146,59 +153,38 @@ class Directory(Base):
     async def copy(
         self, directory: Optional["Directory"], session: SessionContext, config: Config
     ) -> Self:
-        session.add(self)
+        pass
-        await session.refresh(self, attribute_names=["repository"])
-
-        repository_path = await self.repository.path(session, config)
-        directory_path = await self.path(session, config)
-        directory_path = (
-            await directory.path(session, config) if directory else repository_path
-        )
-
-        current_directory = FileSystem(directory_path, repository_path)
-        new_directory = await current_directory.copy(directory_path)
-
-        cloned = self.clone()
-        cloned.name = new_directory.name()
-        cloned.parent_id = directory.id if directory else None
-        session.add(cloned)
-        await session.flush()
-
-        return self
 
     async def move(
         self, directory: Optional["Directory"], session: SessionContext, config: Config
     ) -> Self:
-        session.add(self)
+        pass
-        await session.refresh(self, attribute_names=["repository"])
-
-        repository_path = await self.repository.path(session, config)
-        directory_path = await self.path(session, config)
-        directory_path = (
-            await directory.path(session, config) if directory else repository_path
-        )
-
-        current_directory = FileSystem(directory_path, repository_path)
-        moved_directory = await current_directory.move(directory_path)
-
-        self.name = moved_directory.name()
-        self.parent_id = directory.id if directory else None
-        self.updated = time()
-        await session.flush()
-
-        return self
 
     async def rename(self, name: str, session: SessionContext, config: Config) -> Self:
         session.add(self)
-        await session.refresh(self, attribute_names=["repository"])
 
-        repository_path = await self.repository.path(session, config)
         directory_path = await self.path(session, config)
+        relative_path = await self.relative_path(session)
+        new_path = directory_path.with_name(name)
+        identity = 1
 
-        current_directory = FileSystem(directory_path, repository_path)
+        while True:
-        renamed_directory = await current_directory.rename(name, force=True)
+            if new_path == directory_path:
+                break
+            if not new_path.exists():
+                break
 
-        self.name = renamed_directory.name()
+            new_path = directory_path.with_name(f"{name}.{str(identity)}")
+            identity += 1
+
+        try:
+            await aiofiles.os.rename(directory_path, new_path)
+        except OSError as e:
+            raise DirectoryError(
+                f"Failed to rename directory at /{relative_path}", *e.args
+            )
+
+        self.name = new_path.name
         await session.flush()
         return self
 
@@ -236,4 +222,3 @@ class DirectoryInfo(BaseModel):
 
 from materia.models.repository import Repository
 from materia.models.file import File
-from materia.models.filesystem import FileSystem

src/materia/models/file.py

@@ -47,12 +47,18 @@ class File(Base):
         await session.flush()
         await session.refresh(self, attribute_names=["repository"])
 
+        relative_path = await self.relative_path(session)
         file_path = await self.path(session, config)
+        size = None
 
-        new_file = FileSystem(file_path, await self.repository.path(session, config))
+        try:
-        await new_file.write_file(data)
+            async with aiofiles.open(file_path, mode="wb") as file:
+                await file.write(data)
+            size = (await aiofiles.os.stat(file_path)).st_size
+        except OSError as e:
+            raise FileError(f"Failed to write file at /{relative_path}", *e.args)
 
-        self.size = await new_file.size()
+        self.size = size
         await session.flush()
 
         return self
@@ -60,10 +66,13 @@ class File(Base):
     async def remove(self, session: SessionContext, config: Config):
         session.add(self)
 
+        relative_path = await self.relative_path(session)
         file_path = await self.path(session, config)
 
-        new_file = FileSystem(file_path, await self.repository.path(session, config))
+        try:
-        await new_file.remove()
+            await aiofiles.os.remove(file_path)
+        except OSError as e:
+            raise FileError(f"Failed to remove file at /{relative_path}:", *e.args)
 
         await session.delete(self)
         await session.flush()
@@ -132,25 +141,7 @@ class File(Base):
     async def copy(
         self, directory: Optional["Directory"], session: SessionContext, config: Config
     ) -> Self:
-        session.add(self)
+        pass
-        await session.refresh(self, attribute_names=["repository"])
-
-        repository_path = await self.repository.path(session, config)
-        file_path = await self.path(session, config)
-        directory_path = (
-            await directory.path(session, config) if directory else repository_path
-        )
-
-        current_file = FileSystem(file_path, repository_path)
-        new_file = await current_file.copy(directory_path)
-
-        cloned = self.clone()
-        cloned.name = new_file.name()
-        cloned.parent_id = directory.id if directory else None
-        session.add(cloned)
-        await session.flush()
-
-        return self
 
     async def move(
         self, directory: Optional["Directory"], session: SessionContext, config: Config
@@ -163,29 +154,49 @@ class File(Base):
         directory_path = (
             await directory.path(session, config) if directory else repository_path
         )
+        new_path = File.generate_name(file_path, directory_path, self.name)
 
-        current_file = FileSystem(file_path, repository_path)
+        try:
-        moved_file = await current_file.move(directory_path)
+            await aioshutil.move(file_path, new_path)
+        except OSError as e:
+            raise FileError("Failed to move file:", *e.args)
 
-        self.name = moved_file.name()
         self.parent_id = directory.id if directory else None
-        self.updated = time()
         await session.flush()
 
         return self
 
+    @staticmethod
+    def generate_name(old_file: Path, target_directory: Path, name: str) -> Path:
+        new_path = target_directory.joinpath(name)
+        identity = 1
+
+        while True:
+            if new_path == old_file:
+                break
+            if not new_path.exists():
+                break
+
+            new_path = target_directory.joinpath(
+                f"{name.removesuffix(new_path.suffix)}.{str(identity)}{new_path.suffix}"
+            )
+            identity += 1
+
+        return new_path
+
     async def rename(self, name: str, session: SessionContext, config: Config) -> Self:
         session.add(self)
-        await session.refresh(self, attribute_names=["repository"])
 
-        repository_path = await self.repository.path(session, config)
         file_path = await self.path(session, config)
+        relative_path = await self.relative_path(session)
+        new_path = File.generate_name(file_path, file_path.parent, name)
 
-        current_file = FileSystem(file_path, repository_path)
+        try:
-        renamed_file = await current_file.rename(name, force=True)
+            await aiofiles.os.rename(file_path, new_path)
+        except OSError as e:
+            raise FileError(f"Failed to rename file at /{relative_path}", *e.args)
 
-        self.name = renamed_file.name()
+        self.name = new_path.name
-        self.updated = time()
         await session.flush()
         return self
 
@@ -228,4 +239,3 @@ class FileInfo(BaseModel):
 
 from materia.models.repository import Repository
 from materia.models.directory import Directory
-from materia.models.filesystem import FileSystem

src/materia/models/filesystem.py

@@ -1,186 +0,0 @@
-from typing import Optional, Self, Iterator, TypeVar
-from pathlib import Path
-import aiofiles
-from aiofiles import os as async_os
-from aiofiles import ospath as async_path
-import aioshutil
-import re
-
-
-class FileSystemError(Exception):
-    pass
-
-
-T = TypeVar("T")
-
-
-def wrapped_next(i: Iterator[T]) -> Optional[T]:
-    try:
-        return next(i)
-    except StopIteration:
-        return None
-
-
-class FileSystem:
-    def __init__(self, path: Path, working_directory: Path):
-        if path == Path():
-            raise FileSystemError("The given path is empty")
-        if working_directory == Path():
-            raise FileSystemError("The given working directory is empty")
-
-        self.path = path
-        self.working_directory = working_directory
-        self.relative_path = path.relative_to(working_directory)
-
-    async def exists(self) -> bool:
-        return await async_path.exists(self.path)
-
-    async def size(self) -> int:
-        return await async_path.getsize(self.path)
-
-    async def is_file(self) -> bool:
-        return await async_path.isfile(self.path)
-
-    async def is_directory(self) -> bool:
-        return await async_path.isdir(self.path)
-
-    def name(self) -> str:
-        return self.path.name
-
-    async def remove(self):
-        try:
-            if await self.is_file():
-                await aiofiles.os.remove(self.path)
-
-            if await self.is_directory():
-                await aioshutil.rmtree(str(self.path))
-
-        except OSError as e:
-            raise FileSystemError(
-                f"Failed to remove file system content at /{self.relative_path}:",
-                *e.args,
-            )
-
-    async def generate_name(self, target_directory: Path, name: str) -> str:
-        """Generate name based on target directory contents and self type."""
-        count = 1
-        new_path = target_directory.joinpath(name)
-
-        if new_path == self.path:
-            return name
-
-        while await async_path.exists(new_path):
-            if await self.is_file():
-                if with_counter := re.match(r"^(.+)\.(\d+)\.(\w+)$", new_path.name):
-                    new_name, _, extension = with_counter.groups()
-                elif with_extension := re.match(r"^(.+)\.(\w+)$", new_path.name):
-                    new_name, extension = with_extension.groups()
-
-                new_path = target_directory.joinpath(
-                    "{}.{}.{}".format(new_name, count, extension)
-                )
-
-            if await self.is_directory():
-                if with_counter := re.match(r"^(.+)\.(\d+)$", new_path.name):
-                    new_name, _ = with_counter.groups()
-                else:
-                    new_name = new_path.name
-
-                new_path = target_directory.joinpath("{}.{}".format(new_name, count))
-
-            count += 1
-
-        return new_path.name
-
-    async def move(
-        self,
-        target_directory: Path,
-        new_name: Optional[str] = None,
-        force: bool = False,
-    ):
-        new_name = new_name if new_name else self.path.name
-
-        try:
-            if (
-                await async_path.exists(target_directory.joinpath(new_name))
-                and not force
-            ):
-                raise FileSystemError(
-                    "Failed to move content to target destination: already exists"
-                )
-
-            new_path = target_directory.joinpath(
-                await self.generate_name(target_directory, new_name)
-            )
-            await aioshutil.move(self.path, new_path)
-
-        except Exception as e:
-            raise FileSystemError(
-                f"Failed to move content from /{self.relative_path}:",
-                *e.args,
-            )
-
-        return FileSystem(new_path, self.working_directory)
-
-    async def rename(self, new_name: str, force: bool = False) -> Path:
-        return await self.move(self.path.parent, new_name=new_name, force=force)
-
-    async def copy(
-        self,
-        target_directory: Path,
-        new_name: Optional[str] = None,
-        force: bool = False,
-    ) -> Self:
-        new_name = new_name if new_name else self.path.name
-
-        try:
-            if (
-                await async_path.exists(target_directory.joinpath(new_name))
-                and not force
-            ):
-                raise FileSystemError(
-                    "Failed to copy content to target destination: already exists"
-                )
-
-            new_path = target_directory.joinpath(
-                await self.generate_name(target_directory, new_name)
-            )
-
-            if await self.is_file():
-                await aioshutil.copy(self.path, new_path)
-
-            if await self.is_directory():
-                await aioshutil.copytree(self.path, new_path)
-
-        except Exception as e:
-            raise FileSystemError(
-                f"Failed to copy content from /{self.relative_path}:",
-                *e.args,
-            )
-
-        return FileSystem(new_path, self.working_directory)
-
-    async def make_directory(self):
-        try:
-            if await self.exists():
-                raise FileSystemError("Failed to create directory: already exists")
-
-            await async_os.mkdir(self.path)
-        except Exception as e:
-            raise FileSystemError(
-                f"Failed to create directory at /{self.relative_path}:",
-                *e.args,
-            )
-
-    async def write_file(self, data: bytes):
-        try:
-            if await self.exists():
-                raise FileSystemError("Failed to write file: already exists")
-
-            async with aiofiles.open(self.path, mode="wb") as file:
-                await file.write(data)
-        except Exception as e:
-            raise FileSystemError(
-                f"Failed to write file to /{self.relative_path}:",
-                *e.args,
-            )

src/materia/models/repository.py

@@ -4,7 +4,7 @@ from uuid import UUID, uuid4
 from pathlib import Path
 import shutil
 
-from sqlalchemy import BigInteger, ForeignKey, inspect
+from sqlalchemy import BigInteger, ForeignKey
 from sqlalchemy.orm import mapped_column, Mapped, relationship
 from sqlalchemy.orm.attributes import InstrumentedAttribute
 import sqlalchemy as sa
@@ -86,6 +86,13 @@ class Repository(Base):
         await session.delete(self)
         await session.flush()
 
+    def to_dict(self) -> dict:
+        return {
+            k: getattr(self, k)
+            for k, v in Repository.__dict__.items()
+            if isinstance(v, InstrumentedAttribute)
+        }
+
     async def update(self, session: SessionContext):
         await session.execute(
             sa.update(Repository).values(self.to_dict()).where(Repository.id == self.id)

tests/test_database.py

@@ -181,13 +181,6 @@ async def test_repository(data, tmpdir, session: SessionContext, config: Config)
     assert (await repository.path(session, config)).exists()
     assert await Repository.from_user(data.user, session) == repository
 
-    await session.refresh(repository, attribute_names=["user"])
-    cloned_repository = repository.clone()
-    assert cloned_repository.id is None and cloned_repository.user is None
-    session.add(cloned_repository)
-    await session.flush()
-    assert cloned_repository.id is not None
-
     await repository.remove(session, config)
     make_transient(repository)
     session.add(repository)