diff --git a/alembic/versions/20250414_134126_9f4c8eea_update_route.py b/alembic/versions/20250414_134126_9f4c8eea_update_route.py new file mode 100644 index 0000000..f2b4561 --- /dev/null +++ b/alembic/versions/20250414_134126_9f4c8eea_update_route.py @@ -0,0 +1,33 @@ +"""create table for routes +Revision ID: 2b7a9d4c8f3c +Revises: 0001 +Create Date: 2023-05-24 11:01:46.991122 +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.sql import func +import uuid + +# revision identifiers, used by Alembic. +revision = '2b7a9d4c8f3c' +down_revision = '0001' +branch_labels = None +depends_on = None + +def upgrade(): + op.create_table( + 'routes', + sa.Column('id', sa.String(36), primary_key=True, default=lambda: str(uuid.uuid4())), + sa.Column('name', sa.String(), nullable=False, unique=True), + sa.Column('description', sa.String(), nullable=True), + sa.Column('path', sa.String(), nullable=False, unique=True), + sa.Column('method', sa.String(), nullable=False), + sa.Column('status', sa.Integer(), nullable=False, server_default='1'), + sa.Column('created_at', sa.DateTime(), server_default=func.now()), + sa.Column('updated_at', sa.DateTime(), server_default=func.now(), onupdate=func.now()), + sa.Index('ix_routes_name', 'name'), + sa.Index('ix_routes_path', 'path') + ) + +def downgrade(): + op.drop_table('routes') \ No newline at end of file diff --git a/endpoints/routing.post.py b/endpoints/routing.post.py index e69de29..cbb4912 100644 --- a/endpoints/routing.post.py +++ b/endpoints/routing.post.py @@ -0,0 +1,28 @@ +from typing import List, Optional +from fastapi import APIRouter, Depends +from schemas.route import RouteCreate, RouteSchema +from helpers.route_helpers import create_route, get_routes, validate_unique_route +from sqlalchemy.orm import Session +from core.database import get_db + +router = APIRouter() + +@router.post("/routing", status_code=201, response_model=RouteSchema) +async def create_new_route( + route: RouteCreate, + db: Session = Depends(get_db) +): + if not validate_unique_route(db, route): + return {"detail": "Route with the same name or path already exists"} + + new_route = create_route(db, route) + return new_route + +@router.get("/routing", status_code=200, response_model=List[RouteSchema]) +async def get_all_routes( + name: Optional[str] = None, + path: Optional[str] = None, + db: Session = Depends(get_db) +): + routes = get_routes(db, name, path) + return routes \ No newline at end of file diff --git a/helpers/route_helpers.py b/helpers/route_helpers.py new file mode 100644 index 0000000..c29d028 --- /dev/null +++ b/helpers/route_helpers.py @@ -0,0 +1,114 @@ +import uuid +from typing import List, Optional +from sqlalchemy.orm import Session +from sqlalchemy import or_ + +from models.route import Route +from schemas.route import RouteCreate, RouteUpdate + +def get_route_by_id(db: Session, route_id: uuid.UUID) -> Optional[Route]: + """ + Retrieves a single route by its ID. + + Args: + db (Session): The database session. + route_id (UUID): The ID of the route to retrieve. + + Returns: + Optional[Route]: The route object if found, otherwise None. + """ + return db.query(Route).filter(Route.id == route_id).first() + +def get_routes(db: Session, name: Optional[str] = None, path: Optional[str] = None) -> List[Route]: + """ + Retrieves a list of routes based on optional filters. + + Args: + db (Session): The database session. + name (Optional[str]): The name of the route to filter by. + path (Optional[str]): The path of the route to filter by. + + Returns: + List[Route]: A list of route objects matching the filters. + """ + query = db.query(Route) + if name: + query = query.filter(Route.name.ilike(f"%{name}%")) + if path: + query = query.filter(Route.path.ilike(f"%{path}%")) + return query.all() + +def create_route(db: Session, route_data: RouteCreate) -> Route: + """ + Creates a new route in the database. + + Args: + db (Session): The database session. + route_data (RouteCreate): The data for the route to create. + + Returns: + Route: The newly created route object. + """ + db_route = Route(**route_data.dict()) + db.add(db_route) + db.commit() + db.refresh(db_route) + return db_route + +def update_route(db: Session, route_id: uuid.UUID, route_data: RouteUpdate) -> Optional[Route]: + """ + Updates an existing route in the database. + + Args: + db (Session): The database session. + route_id (UUID): The ID of the route to update. + route_data (RouteUpdate): The updated data for the route. + + Returns: + Optional[Route]: The updated route object if found, otherwise None. + """ + db_route = get_route_by_id(db, route_id) + if not db_route: + return None + + for field, value in route_data.dict(exclude_unset=True).items(): + setattr(db_route, field, value) + + db.commit() + db.refresh(db_route) + return db_route + +def delete_route(db: Session, route_id: uuid.UUID) -> bool: + """ + Deletes a route from the database. + + Args: + db (Session): The database session. + route_id (UUID): The ID of the route to delete. + + Returns: + bool: True if the route was deleted, False otherwise. + """ + db_route = get_route_by_id(db, route_id) + if not db_route: + return False + + db.delete(db_route) + db.commit() + return True + +def validate_unique_route(db: Session, route_data: RouteCreate) -> bool: + """ + Validates if a route with the given name or path already exists. + + Args: + db (Session): The database session. + route_data (RouteCreate): The data for the new route. + + Returns: + bool: True if the route name and path are unique, False otherwise. + """ + existing_route = db.query(Route).filter( + or_(Route.name == route_data.name, Route.path == route_data.path) + ).first() + return existing_route is None \ No newline at end of file diff --git a/models/route.py b/models/route.py new file mode 100644 index 0000000..802defd --- /dev/null +++ b/models/route.py @@ -0,0 +1,17 @@ +from sqlalchemy import Column, String, Integer, DateTime +from sqlalchemy.dialects.postgresql import UUID +from sqlalchemy.sql import func +from core.database import Base +import uuid + +class Route(Base): + __tablename__ = "routes" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + name = Column(String, nullable=False, unique=True, index=True) + description = Column(String, nullable=True) + path = Column(String, nullable=False, unique=True, index=True) + method = Column(String, nullable=False) + status = Column(Integer, nullable=False, default=1) + created_at = Column(DateTime, default=func.now()) + updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) \ No newline at end of file diff --git a/schemas/route.py b/schemas/route.py new file mode 100644 index 0000000..b48d701 --- /dev/null +++ b/schemas/route.py @@ -0,0 +1,29 @@ +from pydantic import BaseModel, Field +from typing import Optional +from datetime import datetime +from uuid import UUID + +class RouteBase(BaseModel): + name: str = Field(..., description="Route name") + description: Optional[str] = Field(None, description="Route description") + path: str = Field(..., description="Route path") + method: str = Field(..., description="Route method") + status: int = Field(..., description="Route status") + +class RouteCreate(RouteBase): + pass + +class RouteUpdate(RouteBase): + name: Optional[str] = Field(None, description="Route name") + description: Optional[str] = Field(None, description="Route description") + path: Optional[str] = Field(None, description="Route path") + method: Optional[str] = Field(None, description="Route method") + status: Optional[int] = Field(None, description="Route status") + +class RouteSchema(RouteBase): + id: UUID + created_at: datetime + updated_at: datetime + + class Config: + orm_mode = True \ No newline at end of file