11 new use command

cloudisk/cli/parser.py

@@ -10,6 +10,7 @@
     link_path,
     list_spaces,
     unlink_path,
+    use_space,
 )
 from cloudisk.http import server
 from cloudisk.vars import CLOUDISK_ROOT
@@ -28,8 +29,8 @@ def init():
 @app.command(help=f"Creates a new space inside '{CLOUDISK_ROOT}'")
 def create(
     name: Annotated[
-        str,
-        typer.Option("--name", "-n", help="Name of the instance."),
+        Optional[str],
+        typer.Argument(help="Name of the instance."),
     ] = None,
     protect: Annotated[
         Optional[bool],
@@ -60,6 +61,16 @@ def create(
     create_space(name, protect)
 
 
+@app.command(help=f"Use a space inside '{CLOUDISK_ROOT}'")
+def use(
+    name: Annotated[
+        str,
+        typer.Argument(help="Name of the instance."),
+    ],
+):
+    use_space(name)
+
+
 @app.command(help="Lists all created spaces")
 def list():
     list_spaces()

cloudisk/db/models/base.py

@@ -1,11 +1,16 @@
 from abc import ABC, abstractmethod
 
 from sqlalchemy import inspect
-from sqlmodel import SQLModel
+from sqlmodel import Field, SQLModel
 
 from cloudisk.globals import context
 
 
+class BaseModel(SQLModel):
+    id: int = Field(primary_key=True)
+    space_id: int = Field(default=None, index=True)
+
+
 class AbstractManager(ABC):
     @abstractmethod
     def table_exists(self) -> bool:
@@ -37,9 +42,11 @@ def table_exists(self) -> bool:
 
 class ModelManager(AbstractManager):
     def __init__(self):  # noqa: D107
-        self.engine = context.engine
+        self.scope = getattr(self.__class__, "scope", context.root)
         self.model = getattr(self.__class__, "model", None)
 
+        self.engine = getattr(self.scope, "engine", None)
+
         SQLModel.metadata.create_all(self.engine)
 
     # Abstract

cloudisk/db/models/group.py

@@ -1,20 +1,17 @@
 from typing import TYPE_CHECKING, Optional
 
-from sqlmodel import Field, Relationship, Session, SQLModel, select
+from sqlmodel import Field, Relationship, Session, select
 
 from cloudisk.db.links import UserGroupLink
-
-from .base import ModelManager
+from cloudisk.db.models.base import BaseModel, ModelManager
 
 if TYPE_CHECKING:
     from cloudisk.db.models.user import UserModel
 
 
-class GroupModel(SQLModel, table=True):
+class GroupModel(BaseModel, table=True):
     __tablename__ = "group"
 
-    id: int = Field(primary_key=True)
-
     name: str = Field(unique=True)
 
     users: list["UserModel"] = Relationship(
@@ -56,7 +53,7 @@ def create(self, name: str) -> GroupModel:
             if session.exec(query).one_or_none():
                 raise Group.AlreadyExists(f"Group '{name}' already exists")
 
-            group = self.model(name=name)
+            group = self.model(space_id=self.scope.extras.get("space_id"), name=name)
 
             session.add(group)
             session.commit()

cloudisk/db/models/metadata.py

@@ -3,17 +3,15 @@
 from typing import Optional
 
 from sqlalchemy.exc import IntegrityError
-from sqlmodel import Field, Session, SQLModel, select
+from sqlmodel import Field, Session, select
 
-from cloudisk.db.models.base import ModelManager
+from cloudisk.db.models.base import BaseModel, ModelManager
 from cloudisk.fs.utils import get_mime_type
 
 
-class MetadataModel(SQLModel, table=True):
+class MetadataModel(BaseModel, table=True):
     __tablename__ = "metadata"
 
-    id: int = Field(primary_key=True)
-
     path: str = Field(unique=True)
     size: int = 0
     content_type: Optional[str] = None
@@ -95,6 +93,7 @@ def create(self, path: Path) -> MetadataModel:
             now = datetime.now()
 
             metadata = self.model(
+                space_id=self.scope.extras.get("space_id"),
                 path=path_str,
                 size=path.stat().st_size,
                 content_type=get_mime_type(path),

cloudisk/db/models/space.py

@@ -10,7 +10,8 @@ class SpaceModel(SQLModel, table=True):
     id: int = Field(primary_key=True)
 
     name: str = Field(unique=True)
-    protect: bool
+    used: bool = False
+    protect: bool = False
 
 
 class Space(ModelManager):
@@ -22,15 +23,15 @@ class Error(Exception):
     class AlreadyExists(Error):  # noqa: N818
         """Raised when the space already exists."""
 
-    def create(self, name: str, protect: bool) -> SpaceModel:
+    def create(self, name: str, protect: bool = False) -> SpaceModel:
         """
         Create a `SpaceModel` instance.
 
         Parameters
         ----------
         name: str
             Name of the space.
-        protect: bool
+        protect: bool = False
             Marks the space as protected with user login.
 
         Returns
@@ -44,10 +45,12 @@ def create(self, name: str, protect: bool) -> SpaceModel:
             When a space already exists.
         """
         with Session(self.engine) as session:
-            space = self.model(
-                name=name,
-                protect=protect,
-            )
+            statement = select(self.model).where(self.model.used)
+            results = session.exec(statement)
+            used_space = results.one_or_none()
+
+            is_used = not used_space
+            space = self.model(name=name, protect=protect, used=is_used)
 
             session.add(space)
 
@@ -60,6 +63,79 @@ def create(self, name: str, protect: bool) -> SpaceModel:
 
             return space
 
+    def remove(self, name: str) -> None:
+        """
+        Remove a `SpaceModel` from the database.
+
+        Parameters
+        ----------
+        name: str
+            Name of the space.
+        """
+        with Session(self.engine) as session:
+            statement = select(self.model).where(self.model.name == name)
+            results = session.exec(statement)
+            space = results.one()
+
+            session.delete(space)
+            session.commit()
+
+    def use(self, name: str) -> SpaceModel:
+        """
+        Set `space.used` to `True`.
+
+        Parameters
+        ----------
+        name: str
+            Name of the space.
+
+        Returns
+        -------
+        SpaceModel
+            The modified instance.
+        """
+        with Session(self.engine) as session:
+            statement = select(self.model).where(self.model.used == 1)
+            results = session.exec(statement)
+            space = results.one_or_none()
+
+            if space:
+                if space.name == name:
+                    return space
+
+                space.used = False
+
+                session.add(space)
+                session.commit()
+
+            statement = select(self.model).where(self.model.name == name)
+            results = session.exec(statement)
+            space = results.one_or_none()
+
+            space.used = True
+
+            session.add(space)
+            session.commit()
+            session.refresh(space)
+
+            return space
+
+    def get_used(self) -> SpaceModel:
+        """
+        Get the space where `space.used` is `True`.
+
+        Returns
+        -------
+        SpaceModel
+            The created instance.
+        """
+        with Session(self.engine) as session:
+            statement = select(self.model).where(self.model.used == 1)
+            results = session.exec(statement)
+            space = results.one_or_none()
+
+            return space
+
     def list(self) -> list[str]:
         """
         List all instances of `SpaceModel`.

cloudisk/db/models/user.py

@@ -3,24 +3,21 @@
 from email.message import EmailMessage
 from typing import TYPE_CHECKING, Optional
 
-from sqlmodel import Field, Relationship, Session, SQLModel, select
+from sqlmodel import Field, Relationship, Session, select
 
 from cloudisk.db.links import UserGroupLink
+from cloudisk.db.models.base import BaseModel, ModelManager
 from cloudisk.globals import settings
 
-from .base import ModelManager
-
 if TYPE_CHECKING:
     from cloudisk.db.models.group import GroupModel
 
 # TODO save encrypted passwords
 
 
-class UserModel(SQLModel, table=True):
+class UserModel(BaseModel, table=True):
     __tablename__ = "user"
 
-    id: int = Field(primary_key=True)
-
     username: str = Field(unique=True)
     email: str = Field(unique=True)
     password: str
@@ -93,6 +90,7 @@ def register(self, username: str, email: str, password: str) -> UserModel:
                 raise User.EmailExists(f"Email '{email}' already exists")
 
             user = self.model(
+                space_id=self.scope.extras.get("space_id"),
                 username=username,
                 email=email,
                 password=password,
@@ -186,7 +184,7 @@ def _send_verify_email(self, email: str):
         msg["To"] = email
 
         if not (email_from := settings.EMAIL_FROM):
-            raise Exception("Please define CLOUDISK_EMAIL_FROM")
+            raise User.Error("Please define EMAIL_FROM")
 
         msg["From"] = email_from
 

cloudisk/fs/commands.py

@@ -3,11 +3,10 @@
 
 import typer
 
-from cloudisk.db.models import Space
 from cloudisk.fs.utils import ask_remove_dir, ask_remove_path
 from cloudisk.logger import get_logger
 from cloudisk.tools.settings import Settings
-from cloudisk.vars import CLOUDISK_DB_FILE, CLOUDISK_ROOT
+from cloudisk.vars import CLOUDISK_DB_FILE, CLOUDISK_ROOT, CLOUDISK_SETTINGS_FILE
 
 logger = get_logger("cloudisk.fs")
 
@@ -108,44 +107,70 @@ def unlink_path(path: Path) -> None:
     logger.info(f"Unlinked '{path}'")
 
 
-def create_space(name: str, protect: bool) -> None:
+def create_space(name: str, protect: bool = False) -> None:
+    from cloudisk.db.models import Space
+
     if not CLOUDISK_ROOT.exists():
         init_cloudisk_root()
 
     space_path = CLOUDISK_ROOT / name
 
-    if space_path.exists() and not ask_remove_dir(space_path):
-        return
+    if space_path.exists():
+        if not ask_remove_dir(space_path):
+            return
+
+        Space().remove(name=name)
 
-    space_path.mkdir()
+    space_path.mkdir(exist_ok=True)
 
-    Settings.build_module(space_path / "settings.py")
+    Settings.build_module(CLOUDISK_ROOT / CLOUDISK_SETTINGS_FILE)
     Space().create(name=name, protect=protect)
 
     logger.info(f"Created the '{name}' space")
 
 
+def use_space(name: str) -> None:
+    from cloudisk.db.models import Space
+
+    if not CLOUDISK_ROOT.exists():
+        init_cloudisk_root()  # pragma: no cover
+
+    space_path = CLOUDISK_ROOT / name
+
+    if not space_path.exists():
+        logger.error(f"Space '{name}' doesn't exist.")
+        return
+
+    Space().use(name=name)
+
+    logger.info(f"Using space '{name}'")
+
+
 # TODO maybe a space is in the database but not found in ROOT
 def list_spaces() -> None:
+    from cloudisk.db.models import Space
+
     spaces = Space().list()
+    used = Space().get_used()
 
     if spaces:
         typer.echo("Tracked spaces:")
         for space in spaces:
-            typer.echo(f"- {space}")
+            if space == used.name:
+                typer.echo(f"+ {space} (used)")
+            else:
+                typer.echo(f"- {space}")
 
     root = os.listdir(CLOUDISK_ROOT)
-    root = [x for x in root if x != CLOUDISK_DB_FILE]
+    root = [x for x in root if x not in [CLOUDISK_DB_FILE, CLOUDISK_SETTINGS_FILE]]
 
-    if len(root):
-        untracked = list(filter(lambda x: x not in spaces, root))
+    untracked = list(filter(lambda x: x not in spaces, root))
 
-        message = "Untracked spaces:"
-        if spaces:
-            message = "\n" + message
+    message = "Untracked spaces:"
+    if spaces:
+        message = "\n" + message
 
+    if untracked:
         typer.echo(message)
         for space in untracked:
             typer.echo(f"- {space}")
-
-        return