feat(routeset): Added route set group creation logic.

zentra_api/cli/commands/add.py

@@ -1,8 +1,18 @@
-import inflect
+import json
+import os
+from typing import Callable
+from pathlib import Path
 
+from zentra_api.cli.conf.checks import zentra_config_path
+from zentra_api.cli.constants import Import, RouteErrorCodes, RouteSuccessCodes
 from zentra_api.cli.builder.routes import RouteBuilder
-from zentra_api.cli.constants.enums import RouteMethods, RouteOptions
-from zentra_api.cli.constants.routes import Name, Route, route_dict_set
+from zentra_api.cli.constants.enums import RouteFile, RouteMethods, RouteOptions
+from zentra_api.cli.constants.routes import Name, Route, route_dict_set, route_imports
+from zentra_api.cli.constants.models import Config, Imports
+
+import inflect
+import typer
+from rich.progress import track
 
 
 def store_name(name: str) -> Name:
@@ -24,36 +34,62 @@ def store_name(name: str) -> Name:
 
 
 def create_api_router(name: str) -> str:
-    """Creates a string representation of an APIRouter."""
-    return f'router = APIRouter(prefix="/{name}", tags=["{name}"])'
+    """
+    Creates a string representation of an APIRouter.
 
+    Parameters:
+        name (str): The name of the route set.
 
-class AddSetOfRoutes:
-    """Performs project operations for the `add-routeset` command."""
+    Returns:
+        str: The string representation of the APIRouter.
+    """
+    return f'router = APIRouter(prefix="/{name.lower()}", tags=["{name.capitalize()}"])'
 
-    def __init__(self, name: str, option: RouteOptions) -> None:
-        self.name = store_name(name.lower().strip())
-        self.option = option
 
-        self.route_map = route_dict_set(self.name)
+def get_route_folder(name: Name, root: Path) -> Path:
+    """
+    Retrieves the path to the route folder.
 
-    def get_routes(self) -> list[Route]:
-        """Retrieves the routes from the route map."""
-        routes = []
+    Parameters:
+        name (Name): The name of the route set.
+        root (Path): The root directory of the project.
 
-        for key, route in self.route_map.items():
-            for letter in self.option:
-                if letter in key:
-                    routes.append(route)
+    Returns:
+        Path: The path to the route folder.
+    """
+    return Path(root, "app", "api", name.plural)
 
-        return routes
+
+class AddSetOfRoutes:
+    """
+    Performs project operations for the `add-routeset` command.
+
+    Parameters:
+        name (str): The name of the route set.
+        option (RouteOptions): The type of route set to create.
+        root (Path): The root directory of the project. Defaults to the parent of the config file. (optional)
+    """
+
+    def __init__(self, name: str, option: RouteOptions, root: Path = None) -> None:
+        self.name = store_name(name.lower().strip())
+        self.root = root if root else zentra_config_path().parent
+        self.route_tasks = AddRouteTasks(name=self.name, root=self.root, option=option)
+
+    def check_folder_exists(self) -> bool:
+        """Checks if the folder name exists in the API directory."""
+        return get_route_folder(self.name, self.root).exists()
 
     def build(self) -> None:
         """Build a set of API routes as models."""
-        routes = self.get_routes()
+        if self.check_folder_exists():
+            raise typer.Exit(code=RouteErrorCodes.FOLDER_EXISTS)
 
-        for route in routes:
-            print(route.to_str())
+        tasks = self.route_tasks.get_tasks_for_set()
+
+        for task in track(tasks, description="Building..."):
+            task()
+
+        raise typer.Exit(code=RouteSuccessCodes.CREATED)
 
 
 class AddRoute:
@@ -95,3 +131,104 @@ def create_delete(self) -> str:
     def build(self) -> None:
         """Builds the route."""
         pass
+
+
+class AddRouteTasks:
+    """Contains the tasks for the `add-routeset` and `add-route` commands."""
+
+    def __init__(self, name: Name, root: Path, option: RouteOptions) -> None:
+        self.name = name
+        self.root = root
+        self.option = option
+
+        self.config: Config = Config(**json.loads(zentra_config_path().read_text()))
+        self.route_map = route_dict_set(self.name)
+        self.api_route_str = create_api_router(self.name.plural)
+        self.route_path = get_route_folder(self.name, self.root)
+
+        self.init_content = None
+
+    def _get_routes(self) -> list[Route]:
+        """Retrieves the routes from the route map."""
+        routes = []
+
+        for key, route in self.route_map.items():
+            for letter in self.option:
+                if letter in key:
+                    routes.append(route)
+
+        return routes
+
+    def _create_init_content(self, routes: list[Route]) -> None:
+        """Creates the '__init__.py' file content."""
+        response_models = [
+            route.response_model for route in routes if route.response_model
+        ]
+        schema_models = [route.schema_model for route in routes if route.schema_model]
+        add_auth = any([route.auth for route in routes])
+
+        folder_imports = [
+            Import(
+                root=".",
+                modules=[RouteFile.RESPONSES.value.split(".")[0]],
+                items=response_models,
+                add_dot=False,
+            ),
+            Import(
+                root=".",
+                modules=[RouteFile.SCHEMA.value.split(".")[0]],
+                items=schema_models,
+                add_dot=False,
+            ),
+        ]
+        file_imports: list[list[Import]] = route_imports(add_auth=add_auth)
+        file_imports.insert(1, folder_imports)
+        file_imports = Imports(items=file_imports).to_str()
+
+        self.init_content = "\n".join(
+            [
+                file_imports,
+                "",
+                self.api_route_str,
+                "\n",
+                "\n".join([route.to_str() + "\n\n" for route in routes]),
+            ]
+        )
+
+    def _update_files(self) -> None:
+        """Updates the '__init__.py', 'schema.py', and 'responses.py' files."""
+        init_file = Path(self.route_path, RouteFile.INIT.value)
+        init_file.write_text(self.init_content)
+
+    def _update_schema_file(self) -> None:
+        """Updates the 'schema.py' file."""
+        pass
+
+    def _update_responses_file(self) -> None:
+        """Updates the 'responses.py' file."""
+        pass
+
+    def _create_route_files(self) -> None:
+        """Creates a new set of route files."""
+        os.makedirs(self.route_path, exist_ok=True)
+
+        for file in RouteFile.values():
+            open(Path(self.route_path, file), "w").close()
+
+    def get_tasks_for_set(self) -> list[Callable]:
+        """Retrieves the tasks for the `add-routeset` command."""
+        tasks = []
+
+        if not self.route_path.exists():
+            tasks.append(self._create_route_files)
+
+        routes = self._get_routes()
+        self._create_init_content(routes)
+
+        tasks.extend([self._update_files])
+
+        return tasks
+
+    def get_tasks_for_route(self) -> list[Callable]:
+        """Retrieves the tasks for the `add-route` command."""
+        pass

zentra_api/cli/constants/__init__.py

@@ -1,5 +1,6 @@
 from enum import Enum
 import importlib.resources as pkg_resources
+from pydantic import BaseModel
 from rich.console import Console
 
 from zentra_api.utils.package import package_path
@@ -59,6 +60,14 @@ class CommonErrorCodes(Enum):
     UNKNOWN_ERROR = 1000
 
 
+class RouteSuccessCodes(Enum):
+    CREATED = 30
+
+
+class RouteErrorCodes(Enum):
+    FOLDER_EXISTS = 40
+
+
 class BuildDetails:
     """A storage container for project build details."""
 
@@ -109,3 +118,61 @@ def __init__(
         "pytest-cov",
     ],
 )
+
+
+class Import(BaseModel):
+    """A model for imports."""
+
+    root: str
+    modules: list[str] | None = None
+    items: list[str]
+    add_dot: bool = True
+
+    def to_str(self) -> str:
+        """Converts the import to a string."""
+        return f"from {self.root}{'.' if self.add_dot else ''}{'.'.join(self.modules) if self.modules else ''} import {", ".join(self.items)}"
+
+
+BASE_IMPORTS = [
+    Import(
+        root="app",
+        modules=["core", "dependencies"],
+        items=["DB_DEPEND"],
+    ),
+    Import(
+        root="app",
+        modules=["db_models"],
+        items=["CONNECT"],
+    ),
+]
+
+AUTH_IMPORTS = [
+    Import(
+        root="app",
+        modules=["auth"],
+        items=["ACTIVE_USER_DEPEND"],
+    )
+]
+
+ZENTRA_IMPORTS = [
+    Import(
+        root="zentra_api",
+        modules=["responses"],
+        items=["SuccessMsgResponse", "get_response_models"],
+    )
+]
+
+FASTAPI_IMPORTS = [
+    Import(
+        root="fastapi",
+        items=["APIRouter", "HTTPException", "status"],
+        add_dot=False,
+    )
+]
+
+
+class RouteImports(Enum):
+    BASE = BASE_IMPORTS
+    AUTH = AUTH_IMPORTS
+    ZENTRA = ZENTRA_IMPORTS
+    FASTAPI = FASTAPI_IMPORTS

zentra_api/cli/constants/message.py

@@ -12,6 +12,8 @@
     MAGIC,
     ROOT_COMMAND,
     CommonErrorCodes,
+    RouteErrorCodes,
+    RouteSuccessCodes,
     SetupSuccessCodes,
 )
 
@@ -37,7 +39,16 @@ def creation_msg(project_name: str, project_type: str, project_path: str) -> str
 """
 
 MISSING_PROJECT = f"""
-Have you run [yellow]{ROOT_COMMAND} init[/yellow]?
+Have you run [yellow]{ROOT_COMMAND} init[/yellow] and are 
+you in the [magenta]project[/magenta] directory?
+"""
+
+ROUTE_CREATED = """
+Access them in the [magenta]api[/magenta] directory.
+"""
+
+ROUTE_SET_EXISTS = """
+You've already created a set of API routes with this name!
 """
 
 UNKNOWN_ERROR = f"""
@@ -66,7 +77,6 @@ def success_msg_with_checks(title: str, desc: str, icon: str = MAGIC) -> str:
     SetupSuccessCodes.ALREADY_CONFIGURED: "",
 }
 
-
 COMMON_ERROR_MAP = {
     CommonErrorCodes.TEST_ERROR: error_msg_with_checks("Test", desc=""),
     CommonErrorCodes.PROJECT_NOT_FOUND: error_msg_with_checks(
@@ -75,10 +85,25 @@ def success_msg_with_checks(title: str, desc: str, icon: str = MAGIC) -> str:
     ),
 }
 
+ROUTE_SUCCESS_MAP = {
+    RouteSuccessCodes.CREATED: success_msg_with_checks(
+        "Route set created!",
+        desc=ROUTE_CREATED,
+    ),
+}
+
+ROUTE_ERROR_MAP = {
+    RouteErrorCodes.FOLDER_EXISTS: error_msg_with_checks(
+        "Route set already exists!",
+        desc=ROUTE_SET_EXISTS,
+    ),
+}
 
 MSG_MAPPER = {
     **SUCCESS_MSG_MAP,
     **COMMON_ERROR_MAP,
+    **ROUTE_SUCCESS_MAP,
+    **ROUTE_ERROR_MAP,
 }
 
 

zentra_api/cli/constants/models.py

@@ -0,0 +1,38 @@
+from pydantic import BaseModel
+
+from zentra_api.cli.constants import Import
+
+
+class APIRouteDetails(BaseModel):
+    """A model for API routes."""
+
+    name: str
+    path: str
+    method: str
+
+
+class Config(BaseModel):
+    """A model for the Zentra config."""
+
+    project_name: str
+    # api_routes: dict[str, APIRouteDetails] | None = None
+
+
+class Imports(BaseModel):
+    """A model for a list of imports."""
+
+    items: list[list[Import]]
+
+    def to_str(self) -> str:
+        """Converts the imports to a string."""
+        imports = []
+
+        for item in self.items:
+            block = []
+            for import_item in item:
+                block.append(import_item.to_str())
+
+            block.append("")
+            imports.append("\n".join(block))
+
+        return "\n".join(imports)