mirror of
https://github.com/cssgunc/compass.git
synced 2025-04-17 17:30:17 -04:00
Implemented API routes for getting all, creating, updating, and deleting resources, services, and tags.
This commit is contained in:
Aidan Kim
parent
a43fb7429a
commit
4c3d49004a
|
|
@ -1,26 +1,43 @@
|
|||
from fastapi import APIRouter, Depends
|
||||
from ..services import ResourceService, UserService
|
||||
from ..models.resource_model import Resource
|
||||
|
||||
from typing import List
|
||||
|
||||
api = APIRouter(prefix="/api/resource")
|
||||
|
||||
openapi_tags = {
|
||||
"name": "Resource",
|
||||
"description": "Resource search and related operations.",
|
||||
}
|
||||
|
||||
|
||||
# TODO: Add security using HTTP Bearer Tokens
|
||||
# TODO: Enable authorization by passing user uuid to API
|
||||
# TODO: Create custom exceptions
|
||||
@api.get("", response_model=List[Resource], tags=["Resource"])
|
||||
def get_all(
|
||||
user_id: str,
|
||||
resource_svc: ResourceService = Depends(),
|
||||
user_svc: UserService = Depends(),
|
||||
):
|
||||
subject = user_svc.get_user_by_uuid(user_id)
|
||||
|
||||
return resource_svc.get_resource_by_user(subject)
|
||||
from fastapi import APIRouter, Depends
|
||||
|
||||
from backend.models.user_model import User
|
||||
from ..services import ResourceService, UserService
|
||||
from ..models.resource_model import Resource
|
||||
|
||||
from typing import List
|
||||
|
||||
api = APIRouter(prefix="/api/resource")
|
||||
|
||||
openapi_tags = {
|
||||
"name": "Resource",
|
||||
"description": "Resource search and related operations.",
|
||||
}
|
||||
|
||||
|
||||
# TODO: Add security using HTTP Bearer Tokens
|
||||
# TODO: Enable authorization by passing user uuid to API
|
||||
# TODO: Create custom exceptions
|
||||
@api.post("", response_model=Resource, tags=["Resource"])
|
||||
def create(
|
||||
subject: User, resource: Resource, resource_svc: ResourceService = Depends()
|
||||
):
|
||||
return resource_svc.create(subject, resource)
|
||||
|
||||
|
||||
@api.get("", response_model=List[Resource], tags=["Resource"])
|
||||
def get_all(subject: User, resource_svc: ResourceService = Depends()):
|
||||
return resource_svc.get_resource_by_user(subject)
|
||||
|
||||
|
||||
@api.put("", response_model=Resource, tags=["Resource"])
|
||||
def update(
|
||||
subject: User, resource: Resource, resource_svc: ResourceService = Depends()
|
||||
):
|
||||
return resource_svc.update(subject, resource)
|
||||
|
||||
|
||||
@api.delete("", response_model=None, tags=["Resource"])
|
||||
def delete(
|
||||
subject: User, resource: Resource, resource_svc: ResourceService = Depends()
|
||||
):
|
||||
resource_svc.delete(subject, resource)
|
||||
|
|
|
|||
|
|
@ -1,26 +1,50 @@
|
|||
from fastapi import APIRouter, Depends
|
||||
from ..services import ServiceService, UserService
|
||||
from ..models.service_model import Service
|
||||
|
||||
from typing import List
|
||||
|
||||
api = APIRouter(prefix="/api/service")
|
||||
|
||||
openapi_tags = {
|
||||
"name": "Service",
|
||||
"description": "Service search and related operations.",
|
||||
}
|
||||
|
||||
|
||||
# TODO: Add security using HTTP Bearer Tokens
|
||||
# TODO: Enable authorization by passing user uuid to API
|
||||
# TODO: Create custom exceptions
|
||||
@api.get("", response_model=List[Service], tags=["Service"])
|
||||
def get_all(
|
||||
user_id: str,
|
||||
service_svc: ServiceService = Depends(),
|
||||
user_svc: UserService = Depends(),
|
||||
):
|
||||
subject = user_svc.get_user_by_uuid(user_id)
|
||||
|
||||
return service_svc.get_service_by_user(subject)
|
||||
from fastapi import APIRouter, Depends
|
||||
|
||||
from backend.models.user_model import User
|
||||
from ..services import ServiceService, UserService
|
||||
from ..models.service_model import Service
|
||||
|
||||
from typing import List
|
||||
|
||||
api = APIRouter(prefix="/api/service")
|
||||
|
||||
openapi_tags = {
|
||||
"name": "Service",
|
||||
"description": "Service search and related operations.",
|
||||
}
|
||||
|
||||
|
||||
# TODO: Add security using HTTP Bearer Tokens
|
||||
# TODO: Enable authorization by passing user uuid to API
|
||||
# TODO: Create custom exceptions
|
||||
@api.post("", response_model=Service, tags=["Service"])
|
||||
def create(
|
||||
subject: User,
|
||||
service: Service,
|
||||
service_svc: ServiceService = Depends()
|
||||
):
|
||||
return service_svc.create(subject, service)
|
||||
|
||||
|
||||
@api.get("", response_model=List[Service], tags=["Service"])
|
||||
def get_all(
|
||||
subject: User,
|
||||
service_svc: ServiceService = Depends()
|
||||
):
|
||||
return service_svc.get_service_by_user(subject)
|
||||
|
||||
@api.put("", response_model=Service, tags=["Service"])
|
||||
def update(
|
||||
subject: User,
|
||||
service: Service,
|
||||
service_svc: ServiceService = Depends()
|
||||
):
|
||||
return service_svc.update(subject, service)
|
||||
|
||||
@api.delete("", response_model=None, tags=["Service"])
|
||||
def delete(
|
||||
subject: User,
|
||||
service: Service,
|
||||
service_svc: ServiceService = Depends()
|
||||
):
|
||||
service_svc.delete(subject, service)
|
||||
|
|
|
|||
51
backend/api/tag.py
Normal file
51
backend/api/tag.py
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
from fastapi import APIRouter, Depends
|
||||
|
||||
from backend.models.tag_model import Tag
|
||||
from backend.models.user_model import User
|
||||
from backend.services.tag import TagService
|
||||
from ..services import ResourceService, UserService
|
||||
from ..models.resource_model import Resource
|
||||
|
||||
from typing import List
|
||||
|
||||
api = APIRouter(prefix="/api/tag")
|
||||
|
||||
openapi_tags = {
|
||||
"name": "Tag",
|
||||
"description": "Tag CRUD operations.",
|
||||
}
|
||||
|
||||
|
||||
# TODO: Add security using HTTP Bearer Tokens
|
||||
# TODO: Enable authorization by passing user uuid to API
|
||||
# TODO: Create custom exceptions
|
||||
@api.post("", response_model=Tag, tags=["Tag"])
|
||||
def create(
|
||||
subject: User,
|
||||
tag: Tag,
|
||||
tag_service: TagService
|
||||
):
|
||||
return tag_service.create(subject, tag)
|
||||
|
||||
@api.get("", response_model=List[Tag], tags=["Tag"])
|
||||
def get_all(
|
||||
subject: User,
|
||||
tag_svc: TagService
|
||||
):
|
||||
return tag_svc.get_all()
|
||||
|
||||
@api.put("", response_model=Tag, tags=["Tag"])
|
||||
def delete(
|
||||
subject: User,
|
||||
tag: Tag,
|
||||
tag_svc: TagService
|
||||
):
|
||||
return tag_svc.delete(subject, tag)
|
||||
|
||||
@api.delete("", response_model=None, tags=["Tag"])
|
||||
def delete(
|
||||
subject: User,
|
||||
tag: Tag,
|
||||
tag_svc: TagService
|
||||
):
|
||||
tag_svc.delete(subject, tag)
|
||||
|
|
@ -1,67 +1,65 @@
|
|||
""" Defines the table for storing resources """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import Integer, String, DateTime, Enum
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
# Import datetime for created_at type
|
||||
from datetime import datetime
|
||||
|
||||
# Import self for to model
|
||||
from typing import Self
|
||||
from backend.entities.program_enum import Program_Enum
|
||||
from ..models.resource_model import Resource
|
||||
|
||||
|
||||
class ResourceEntity(EntityBase):
|
||||
|
||||
# set table name
|
||||
__tablename__ = "resource"
|
||||
|
||||
# set fields
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
|
||||
name: Mapped[str] = mapped_column(String(64), nullable=False)
|
||||
summary: Mapped[str] = mapped_column(String(100), nullable=False)
|
||||
link: Mapped[str] = mapped_column(String, nullable=False)
|
||||
program: Mapped[Program_Enum] = mapped_column(Enum(Program_Enum), nullable=False)
|
||||
# relationships
|
||||
resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(
|
||||
back_populates="resource", cascade="all,delete"
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model: Resource) -> Self:
|
||||
"""
|
||||
Create a UserEntity from a User model.
|
||||
|
||||
Args:
|
||||
model (User): The model to create the entity from.
|
||||
|
||||
Returns:
|
||||
Self: The entity (not yet persisted).
|
||||
"""
|
||||
|
||||
return cls(
|
||||
id=model.id,
|
||||
created_at=model.created_at,
|
||||
name=model.name,
|
||||
summary=model.summary,
|
||||
link=model.link,
|
||||
program=model.program,
|
||||
)
|
||||
|
||||
def to_model(self) -> Resource:
|
||||
return Resource(
|
||||
id=self.id,
|
||||
created_at=self.created_at,
|
||||
name=self.name,
|
||||
summary=self.summary,
|
||||
link=self.link,
|
||||
program=self.program,
|
||||
)
|
||||
""" Defines the table for storing resources """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import ForeignKey, Integer, String, DateTime, Enum
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
# Import datetime for created_at type
|
||||
from datetime import datetime
|
||||
|
||||
# Import self for to model
|
||||
from typing import Self
|
||||
from backend.entities.program_enum import Program_Enum
|
||||
from ..models.resource_model import Resource
|
||||
|
||||
|
||||
class ResourceEntity(EntityBase):
|
||||
|
||||
# set table name
|
||||
__tablename__ = "resource"
|
||||
|
||||
# set fields
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
|
||||
name: Mapped[str] = mapped_column(String(64), nullable=False)
|
||||
summary: Mapped[str] = mapped_column(String(100), nullable=False)
|
||||
link: Mapped[str] = mapped_column(String, nullable=False)
|
||||
program: Mapped[Program_Enum] = mapped_column(Enum(Program_Enum), nullable=False)
|
||||
# relationships
|
||||
tags: Mapped[list["ResourceTagEntity"]] = relationship(back_populates="resource")
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model: Resource) -> Self:
|
||||
"""
|
||||
Create a UserEntity from a User model.
|
||||
|
||||
Args:
|
||||
model (User): The model to create the entity from.
|
||||
|
||||
Returns:
|
||||
Self: The entity (not yet persisted).
|
||||
"""
|
||||
|
||||
return cls(
|
||||
id=model.id,
|
||||
created_at=model.created_at,
|
||||
name=model.name,
|
||||
summary=model.summary,
|
||||
link=model.link,
|
||||
program=model.program,
|
||||
)
|
||||
|
||||
def to_model(self) -> Resource:
|
||||
return Resource(
|
||||
id=self.id,
|
||||
created_at=self.created_at,
|
||||
name=self.name,
|
||||
summary=self.summary,
|
||||
link=self.link,
|
||||
program=self.program,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,46 +1,44 @@
|
|||
""" Defines the table for resource tags """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import ForeignKey, Integer, String, DateTime
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
# Import datetime for created_at type
|
||||
from datetime import datetime
|
||||
|
||||
# Import self for to model
|
||||
from typing import Self
|
||||
|
||||
|
||||
class ResourceTagEntity(EntityBase):
|
||||
|
||||
# set table name to user in the database
|
||||
__tablename__ = "resource_tag"
|
||||
|
||||
# set fields or 'columns' for the user table
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
resourceId: Mapped[int] = mapped_column(ForeignKey("resource.id"))
|
||||
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
|
||||
|
||||
# relationships
|
||||
resource: Mapped["ResourceEntity"] = relationship(back_populates="resourceTags")
|
||||
tag: Mapped["TagEntity"] = relationship(back_populates="resourceTags")
|
||||
|
||||
# @classmethod
|
||||
# def from_model (cls, model: resource_tag_model) -> Self:
|
||||
# return cls (
|
||||
# id = model.id,
|
||||
# resourceId = model.resourceId,
|
||||
# tagId = model.tagId,
|
||||
# )
|
||||
|
||||
# def to_model (self) -> resource_tag_model:
|
||||
# return user_model(
|
||||
# id = self.id,
|
||||
# resourceId = self.resourceId,
|
||||
# tagId = self.tagId,
|
||||
# )
|
||||
""" Defines the table for resource tags """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import ForeignKey, Integer, String, DateTime
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
# Import datetime for created_at type
|
||||
from datetime import datetime
|
||||
|
||||
# Import self for to model
|
||||
from typing import Self
|
||||
|
||||
|
||||
class ResourceTagEntity(EntityBase):
|
||||
|
||||
# set table name to user in the database
|
||||
__tablename__ = "resource_tag"
|
||||
|
||||
# set fields or 'columns' for the user table
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
resource_id: Mapped[int] = mapped_column(ForeignKey("resource.id"), primary_key=True)
|
||||
tag_id: Mapped[int] = mapped_column(ForeignKey("tag.id", primary_key=True))
|
||||
resource: Mapped["ResourceEntity"] = mapped_column(backpopulates="tags")
|
||||
tag: Mapped["TagEntity"] = mapped_column(backpopulates="resource_tags")
|
||||
|
||||
# @classmethod
|
||||
# def from_model (cls, model: resource_tag_model) -> Self:
|
||||
# return cls (
|
||||
# id = model.id,
|
||||
# resourceId = model.resourceId,
|
||||
# tagId = model.tagId,
|
||||
# )
|
||||
|
||||
# def to_model (self) -> resource_tag_model:
|
||||
# return user_model(
|
||||
# id = self.id,
|
||||
# resourceId = self.resourceId,
|
||||
# tagId = self.tagId,
|
||||
# )
|
||||
|
|
|
|||
|
|
@ -1,47 +1,64 @@
|
|||
""" Defines the table for storing services """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import Integer, String, DateTime, ARRAY
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
# Import datetime for created_at type
|
||||
from datetime import datetime
|
||||
|
||||
# Import enums for Program
|
||||
import enum
|
||||
from sqlalchemy import Enum
|
||||
|
||||
from backend.models.service_model import Service
|
||||
from typing import Self
|
||||
from backend.models.enum_for_models import ProgramTypeEnum
|
||||
|
||||
class ServiceEntity(EntityBase):
|
||||
|
||||
# set table name
|
||||
__tablename__ = "service"
|
||||
|
||||
# set fields
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
|
||||
name: Mapped[str] = mapped_column(String(32), nullable=False)
|
||||
status: Mapped[str] = mapped_column(String(32), nullable=False)
|
||||
summary: Mapped[str] = mapped_column(String(100), nullable=False)
|
||||
requirements: Mapped[list[str]] = mapped_column(ARRAY(String))
|
||||
program: Mapped[ProgramTypeEnum] = mapped_column(Enum(ProgramTypeEnum), nullable=False)
|
||||
|
||||
# relationships
|
||||
serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(
|
||||
back_populates="service", cascade="all,delete"
|
||||
)
|
||||
|
||||
def to_model(self) -> Service:
|
||||
return Service(id=self.id, name=self.name, status=self.status, summary=self.summary, requirements=self.requirements, program=self.program)
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model:Service) -> Self:
|
||||
return cls(id=model.id, name=model.name, status=model.status, summary=model.summary, requirements=model.requirements, program=model.program)
|
||||
""" Defines the table for storing services """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import Integer, String, DateTime, ARRAY
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
# Import datetime for created_at type
|
||||
from datetime import datetime
|
||||
|
||||
# Import enums for Program
|
||||
import enum
|
||||
from sqlalchemy import Enum
|
||||
|
||||
from backend.models.service_model import Service
|
||||
from typing import Self
|
||||
from backend.models.enum_for_models import ProgramTypeEnum
|
||||
|
||||
|
||||
class ServiceEntity(EntityBase):
|
||||
|
||||
# set table name
|
||||
__tablename__ = "service"
|
||||
|
||||
# set fields
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
|
||||
name: Mapped[str] = mapped_column(String(32), nullable=False)
|
||||
status: Mapped[str] = mapped_column(String(32), nullable=False)
|
||||
summary: Mapped[str] = mapped_column(String(100), nullable=False)
|
||||
requirements: Mapped[list[str]] = mapped_column(ARRAY(String))
|
||||
program: Mapped[ProgramTypeEnum] = mapped_column(
|
||||
Enum(ProgramTypeEnum), nullable=False
|
||||
)
|
||||
|
||||
# relationships
|
||||
tags: Mapped[list["ServiceTagEntity"]] = relationship(
|
||||
back_populates="service", cascade="all,delete"
|
||||
)
|
||||
|
||||
def to_model(self) -> Service:
|
||||
return Service(
|
||||
id=self.id,
|
||||
name=self.name,
|
||||
status=self.status,
|
||||
summary=self.summary,
|
||||
requirements=self.requirements,
|
||||
program=self.program,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model: Service) -> Self:
|
||||
return cls(
|
||||
id=model.id,
|
||||
name=model.name,
|
||||
status=model.status,
|
||||
summary=model.summary,
|
||||
requirements=model.requirements,
|
||||
program=model.program,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,25 +1,23 @@
|
|||
""" Defines the table for service tags """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import ForeignKey, Integer
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
|
||||
class ServiceTagEntity(EntityBase):
|
||||
|
||||
# set table name to user in the database
|
||||
__tablename__ = "service_tag"
|
||||
|
||||
# set fields or 'columns' for the user table
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
serviceId: Mapped[int] = mapped_column(ForeignKey("service.id"))
|
||||
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
|
||||
|
||||
# relationships
|
||||
service: Mapped["ServiceEntity"] = relationship(back_populates="serviceTags")
|
||||
tag: Mapped["TagEntity"] = relationship(back_populates="serviceTags")
|
||||
""" Defines the table for service tags """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import ForeignKey, Integer
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
|
||||
class ServiceTagEntity(EntityBase):
|
||||
|
||||
# set table name to user in the database
|
||||
__tablename__ = "service_tag"
|
||||
|
||||
# set fields or 'columns' for the user table
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
resource_id: Mapped[int] = mapped_column(ForeignKey("service.id"), primary_key=True)
|
||||
tag_id: Mapped[int] = mapped_column(ForeignKey("tag.id", primary_key=True))
|
||||
service: Mapped["ServiceEntity"] = mapped_column(backpopulates="tags")
|
||||
tag: Mapped["TagEntity"] = mapped_column(backpopulates="service_tags")
|
||||
|
|
|
|||
|
|
@ -1,65 +1,63 @@
|
|||
""" Defines the table for storing tags """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import Integer, String, DateTime
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
# Import datetime for created_at type
|
||||
from datetime import datetime
|
||||
|
||||
from ..models.tag_model import Tag
|
||||
|
||||
from typing import Self
|
||||
|
||||
class TagEntity(EntityBase):
|
||||
|
||||
#set table name
|
||||
__tablename__ = "tag"
|
||||
|
||||
#set fields
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
|
||||
content: Mapped[str] = mapped_column(String(100), nullable=False)
|
||||
|
||||
#relationships
|
||||
resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(back_populates="tag", cascade="all,delete")
|
||||
serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(back_populates="tag", cascade="all,delete")
|
||||
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model: Tag) -> Self:
|
||||
"""
|
||||
Create a user entity from model
|
||||
|
||||
Args: model (User): the model to create the entity from
|
||||
|
||||
Returns:
|
||||
self: The entity
|
||||
"""
|
||||
|
||||
return cls(
|
||||
id=model.id,
|
||||
content=model.id,
|
||||
)
|
||||
|
||||
def to_model(self) -> Tag:
|
||||
"""
|
||||
Create a user model from entity
|
||||
|
||||
Returns:
|
||||
User: A User model for API usage
|
||||
"""
|
||||
|
||||
return Tag(
|
||||
id=self.id,
|
||||
content=self.content,
|
||||
)
|
||||
|
||||
|
||||
|
||||
""" Defines the table for storing tags """
|
||||
|
||||
# Import our mapped SQL types from SQLAlchemy
|
||||
from sqlalchemy import Integer, String, DateTime
|
||||
|
||||
# Import mapping capabilities from the SQLAlchemy ORM
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
# Import the EntityBase that we are extending
|
||||
from .entity_base import EntityBase
|
||||
|
||||
# Import datetime for created_at type
|
||||
from datetime import datetime
|
||||
|
||||
from ..models.tag_model import Tag
|
||||
|
||||
from typing import Self
|
||||
|
||||
|
||||
class TagEntity(EntityBase):
|
||||
|
||||
# set table name
|
||||
__tablename__ = "tag"
|
||||
|
||||
# set fields
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
|
||||
content: Mapped[str] = mapped_column(String(100), nullable=False)
|
||||
resource_tags: Mapped[list["ResourceTagEntity"]] = relationship(
|
||||
back_populates="tag"
|
||||
)
|
||||
service_tags: Mapped[list["ServiceTagEntity"]] = relationship(back_populates="tag")
|
||||
|
||||
@classmethod
|
||||
def from_model(cls, model: Tag) -> Self:
|
||||
"""
|
||||
Create a user entity from model
|
||||
|
||||
Args: model (User): the model to create the entity from
|
||||
|
||||
Returns:
|
||||
self: The entity
|
||||
"""
|
||||
|
||||
return cls(
|
||||
id=model.id,
|
||||
content=model.id,
|
||||
created_at=model.created_at
|
||||
)
|
||||
|
||||
def to_model(self) -> Tag:
|
||||
"""
|
||||
Create a user model from entity
|
||||
|
||||
Returns:
|
||||
User: A User model for API usage
|
||||
"""
|
||||
|
||||
return Tag(
|
||||
id=self.id,
|
||||
content=self.content,
|
||||
created_at=self.created_at
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,165 +1,160 @@
|
|||
from fastapi import Depends
|
||||
from ..database import db_session
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import select
|
||||
from ..models.resource_model import Resource
|
||||
from ..entities.resource_entity import ResourceEntity
|
||||
from ..models.user_model import User, UserTypeEnum
|
||||
|
||||
from .exceptions import ResourceNotFoundException
|
||||
|
||||
|
||||
class ResourceService:
|
||||
|
||||
def __init__(self, session: Session = Depends(db_session)):
|
||||
self._session = session
|
||||
|
||||
def get_resource_by_user(self, subject: User):
|
||||
"""Resource method getting all of the resources that a user has access to based on role"""
|
||||
if subject.role != UserTypeEnum.VOLUNTEER:
|
||||
query = select(ResourceEntity)
|
||||
entities = self._session.scalars(query).all()
|
||||
|
||||
return [resource.to_model() for resource in entities]
|
||||
else:
|
||||
programs = subject.program
|
||||
resources = []
|
||||
for program in programs:
|
||||
query = select(ResourceEntity).filter(ResourceEntity.program == program)
|
||||
entities = self._session.scalars(query).all()
|
||||
for entity in entities:
|
||||
resources.append(entity)
|
||||
|
||||
return [resource.to_model() for resource in resources]
|
||||
|
||||
def create(self, user: User, resource: Resource) -> Resource:
|
||||
"""
|
||||
Creates a resource based on the input object and adds it to the table if the user has the right permissions.
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
resource: Resource object to add to table
|
||||
|
||||
Returns:
|
||||
Resource: Object added to table
|
||||
"""
|
||||
if resource.role != user.role or resource.group != user.group:
|
||||
raise PermissionError(
|
||||
"User does not have permission to add resources in this role or group."
|
||||
)
|
||||
|
||||
resource_entity = ResourceEntity.from_model(resource)
|
||||
self._session.add(resource_entity)
|
||||
self._session.commit()
|
||||
|
||||
return resource_entity.to_model()
|
||||
|
||||
def get_by_id(self, user: User, id: int) -> Resource:
|
||||
"""
|
||||
Gets a resource based on the resource id that the user has access to
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
id: int, the id of the resource
|
||||
|
||||
Returns:
|
||||
Resource
|
||||
|
||||
Raises:
|
||||
ResourceNotFoundException: If no resource is found with id
|
||||
"""
|
||||
resource = (
|
||||
self._session.query(ResourceEntity)
|
||||
.filter(
|
||||
ResourceEntity.id == id,
|
||||
ResourceEntity.role == user.role,
|
||||
ResourceEntity.group == user.group,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
|
||||
if resource is None:
|
||||
raise ResourceNotFoundException(f"No resource found with id: {id}")
|
||||
|
||||
return resource.to_model()
|
||||
|
||||
def update(self, user: User, resource: ResourceEntity) -> Resource:
|
||||
"""
|
||||
Update the resource if the user has access
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
resource (ResourceEntity): Resource to update
|
||||
|
||||
Returns:
|
||||
Resource: Updated resource object
|
||||
|
||||
Raises:
|
||||
ResourceNotFoundException: If no resource is found with the corresponding ID
|
||||
"""
|
||||
if resource.role != user.role or resource.group != user.group:
|
||||
raise PermissionError(
|
||||
"User does not have permission to update this resource."
|
||||
)
|
||||
|
||||
obj = self._session.get(ResourceEntity, resource.id) if resource.id else None
|
||||
|
||||
if obj is None:
|
||||
raise ResourceNotFoundException(
|
||||
f"No resource found with matching id: {resource.id}"
|
||||
)
|
||||
|
||||
obj.update_from_model(resource) # Assuming an update method exists
|
||||
self._session.commit()
|
||||
|
||||
return obj.to_model()
|
||||
|
||||
def delete(self, user: User, id: int) -> None:
|
||||
"""
|
||||
Delete resource based on id that the user has access to
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
id: int, a unique resource id
|
||||
|
||||
Raises:
|
||||
ResourceNotFoundException: If no resource is found with the corresponding id
|
||||
"""
|
||||
resource = (
|
||||
self._session.query(ResourceEntity)
|
||||
.filter(
|
||||
ResourceEntity.id == id,
|
||||
ResourceEntity.role == user.role,
|
||||
ResourceEntity.group == user.group,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
|
||||
if resource is None:
|
||||
raise ResourceNotFoundException(f"No resource found with matching id: {id}")
|
||||
|
||||
self._session.delete(resource)
|
||||
self._session.commit()
|
||||
|
||||
def get_by_slug(self, user: User, search_string: str) -> list[Resource]:
|
||||
"""
|
||||
Get a list of resources given a search string that the user has access to
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
search_string: a string to search resources by
|
||||
|
||||
Returns:
|
||||
list[Resource]: list of resources relating to the string
|
||||
|
||||
Raises:
|
||||
ResourceNotFoundException if no resource is found with the corresponding slug
|
||||
"""
|
||||
query = select(ResourceEntity).where(
|
||||
ResourceEntity.title.ilike(f"%{search_string}%"),
|
||||
ResourceEntity.role == user.role,
|
||||
ResourceEntity.group == user.group,
|
||||
)
|
||||
entities = self._session.scalars(query).all()
|
||||
|
||||
return [entity.to_model() for entity in entities]
|
||||
from fastapi import Depends
|
||||
from ..database import db_session
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import select
|
||||
from ..models.resource_model import Resource
|
||||
from ..entities.resource_entity import ResourceEntity
|
||||
from ..models.user_model import User, UserTypeEnum
|
||||
|
||||
from .exceptions import ResourceNotFoundException
|
||||
|
||||
|
||||
class ResourceService:
|
||||
|
||||
def __init__(self, session: Session = Depends(db_session)):
|
||||
self._session = session
|
||||
|
||||
def get_resource_by_user(self, subject: User) -> list[Resource]:
|
||||
"""Resource method getting all of the resources that a user has access to based on role"""
|
||||
if subject.role != UserTypeEnum.VOLUNTEER:
|
||||
query = select(ResourceEntity)
|
||||
entities = self._session.scalars(query).all()
|
||||
return [resource.to_model() for resource in entities]
|
||||
else:
|
||||
programs = subject.program
|
||||
resources = []
|
||||
for program in programs:
|
||||
entities = self._session.query(ResourceEntity).where(ResourceEntity.program == program).all()
|
||||
for entity in entities:
|
||||
resources.append(entity.to_model())
|
||||
return [resource for resource in resources]
|
||||
|
||||
def create(self, subject: User, resource: Resource) -> Resource:
|
||||
"""
|
||||
Creates a resource based on the input object and adds it to the table if the user has the right permissions.
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
resource: Resource object to add to table
|
||||
|
||||
Returns:
|
||||
Resource: Object added to table
|
||||
"""
|
||||
# Ask about what the requirements for making a resource are.
|
||||
if resource.role != subject.role or resource.group != subject.group:
|
||||
raise PermissionError(
|
||||
"User does not have permission to add resources in this role or group."
|
||||
)
|
||||
|
||||
resource_entity = ResourceEntity.from_model(resource)
|
||||
self._session.add(resource_entity)
|
||||
self._session.commit()
|
||||
|
||||
return resource_entity.to_model()
|
||||
|
||||
def get_by_id(self, user: User, id: int) -> Resource:
|
||||
"""
|
||||
Gets a resource based on the resource id that the user has access to
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
id: int, the id of the resource
|
||||
|
||||
Returns:
|
||||
Resource
|
||||
|
||||
Raises:
|
||||
ResourceNotFoundException: If no resource is found with id
|
||||
"""
|
||||
resource = (
|
||||
self._session.query(ResourceEntity)
|
||||
.filter(
|
||||
ResourceEntity.id == id,
|
||||
ResourceEntity.role == user.role,
|
||||
ResourceEntity.group == user.group,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
|
||||
if resource is None:
|
||||
raise ResourceNotFoundException(f"No resource found with id: {id}")
|
||||
|
||||
return resource.to_model()
|
||||
|
||||
def update(self, subject: User, resource: Resource) -> Resource:
|
||||
"""
|
||||
Update the resource if the user has access
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
resource (ResourceEntity): Resource to update
|
||||
|
||||
Returns:
|
||||
Resource: Updated resource object
|
||||
|
||||
Raises:
|
||||
ResourceNotFoundException: If no resource is found with the corresponding ID
|
||||
"""
|
||||
if resource.role != subject.role or resource.group != subject.group:
|
||||
raise PermissionError(
|
||||
"User does not have permission to update this resource."
|
||||
)
|
||||
|
||||
query = select(ResourceEntity).where(ResourceEntity.id == resource.id)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise ResourceNotFoundException(
|
||||
f"No resource found with matching id: {resource.id}"
|
||||
)
|
||||
|
||||
entity.name = resource.name
|
||||
entity.summary = resource.summary
|
||||
entity.link = resource.link
|
||||
entity.program = resource.program
|
||||
self._session.commit()
|
||||
|
||||
return entity.to_model()
|
||||
|
||||
def delete(self, subject: User, resource: Resource) -> None:
|
||||
"""
|
||||
Delete resource based on id that the user has access to
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
id: int, a unique resource id
|
||||
|
||||
Raises:
|
||||
ResourceNotFoundException: If no resource is found with the corresponding id
|
||||
"""
|
||||
query = select(ResourceEntity).where(ResourceEntity.id == resource.id)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise ResourceNotFoundException(f"No resource found with matching id: {resource.id}")
|
||||
|
||||
self._session.delete(entity)
|
||||
self._session.commit()
|
||||
|
||||
def get_by_slug(self, user: User, search_string: str) -> list[Resource]:
|
||||
"""
|
||||
Get a list of resources given a search string that the user has access to
|
||||
|
||||
Parameters:
|
||||
user: a valid User model representing the currently logged in User
|
||||
search_string: a string to search resources by
|
||||
|
||||
Returns:
|
||||
list[Resource]: list of resources relating to the string
|
||||
|
||||
Raises:
|
||||
ResourceNotFoundException if no resource is found with the corresponding slug
|
||||
"""
|
||||
query = select(ResourceEntity).where(
|
||||
ResourceEntity.title.ilike(f"%{search_string}%"),
|
||||
ResourceEntity.role == user.role,
|
||||
ResourceEntity.group == user.group,
|
||||
)
|
||||
entities = self._session.scalars(query).all()
|
||||
|
||||
return [entity.to_model() for entity in entities]
|
||||
|
|
|
|||
|
|
@ -1,127 +1,114 @@
|
|||
from fastapi import Depends
|
||||
|
||||
from ..database import db_session
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import func, select, and_, func, or_, exists, or_
|
||||
|
||||
from backend.models.service_model import Service
|
||||
from backend.models.user_model import User
|
||||
from backend.entities.service_entity import ServiceEntity
|
||||
from backend.models.enum_for_models import ProgramTypeEnum, UserTypeEnum
|
||||
from backend.services.exceptions import (
|
||||
ServiceNotFoundException,
|
||||
ProgramNotAssignedException,
|
||||
)
|
||||
|
||||
|
||||
class ServiceService:
|
||||
|
||||
def __init__(self, session: Session = Depends(db_session)):
|
||||
self._session = session
|
||||
|
||||
def get_service_by_program(self, program: ProgramTypeEnum) -> list[Service]:
|
||||
"""Service method getting services belonging to a particular program."""
|
||||
query = select(ServiceEntity).filter(ServiceEntity.program == program)
|
||||
entities = self._session.scalars(query)
|
||||
|
||||
return [entity.to_model() for entity in entities]
|
||||
|
||||
def get_service_by_id(self, id: int) -> Service:
|
||||
"""Service method getting services by id."""
|
||||
query = select(ServiceEntity).filter(ServiceEntity.id == id)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise ServiceNotFoundException(f"Service with id: {id} does not exist")
|
||||
|
||||
return entity.to_model()
|
||||
|
||||
def get_service_by_name(self, name: str) -> Service:
|
||||
"""Service method getting services by id."""
|
||||
query = select(ServiceEntity).filter(ServiceEntity.name == name)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise ServiceNotFoundException(f"Service with name: {name} does not exist")
|
||||
|
||||
return entity.to_model()
|
||||
|
||||
def get_service_by_user(self, subject: User):
|
||||
"""Service method getting all of the services that a user has access to based on role"""
|
||||
if subject.role != UserTypeEnum.VOLUNTEER:
|
||||
query = select(ServiceEntity)
|
||||
entities = self._session.scalars(query).all()
|
||||
|
||||
return [service.to_model() for service in entities]
|
||||
else:
|
||||
programs = subject.program
|
||||
services = []
|
||||
for program in programs:
|
||||
query = select(ServiceEntity).filter(ServiceEntity.program == program)
|
||||
entities = self._session.scalars(query).all()
|
||||
for entity in entities:
|
||||
services.append(entity)
|
||||
|
||||
return [service.to_model() for service in services]
|
||||
|
||||
def get_all(self, subject: User) -> list[Service]:
|
||||
"""Service method retrieving all of the services in the table."""
|
||||
if subject.role == UserTypeEnum.VOLUNTEER:
|
||||
raise ProgramNotAssignedException(
|
||||
f"User is not {UserTypeEnum.ADMIN} or {UserTypeEnum.VOLUNTEER}, cannot get all"
|
||||
)
|
||||
|
||||
query = select(ServiceEntity)
|
||||
entities = self._session.scalars(query).all()
|
||||
|
||||
return [service.to_model() for service in entities]
|
||||
|
||||
def create(self, subject: User, service: Service) -> Service:
|
||||
"""Creates/adds a service to the table."""
|
||||
if subject.role != UserTypeEnum.ADMIN:
|
||||
raise ProgramNotAssignedException(
|
||||
f"User is not {UserTypeEnum.ADMIN}, cannot create service"
|
||||
)
|
||||
|
||||
service_entity = ServiceEntity.from_model(service)
|
||||
self._session.add(service_entity)
|
||||
self._session.commit()
|
||||
return service_entity.to_model()
|
||||
|
||||
def update(self, subject: User, service: Service) -> Service:
|
||||
"""Updates a service if in the table."""
|
||||
if subject.role != UserTypeEnum.ADMIN:
|
||||
raise ProgramNotAssignedException(
|
||||
f"User is not {UserTypeEnum.ADMIN}, cannot update service"
|
||||
)
|
||||
|
||||
service_entity = self._session.get(ServiceEntity, service.id)
|
||||
|
||||
if service_entity is None:
|
||||
raise ServiceNotFoundException(
|
||||
"The service you are searching for does not exist."
|
||||
)
|
||||
|
||||
service_entity.name = service.name
|
||||
service_entity.status = service.status
|
||||
service_entity.summary = service.summary
|
||||
service_entity.requirements = service.requirements
|
||||
service_entity.program = service.program
|
||||
|
||||
self._session.commit()
|
||||
|
||||
return service_entity.to_model()
|
||||
|
||||
def delete(self, subject: User, service: Service) -> None:
|
||||
"""Deletes a service from the table."""
|
||||
if subject.role != UserTypeEnum.ADMIN:
|
||||
raise ProgramNotAssignedException(f"User is not {UserTypeEnum.ADMIN}")
|
||||
service_entity = self._session.get(ServiceEntity, service.id)
|
||||
|
||||
if service_entity is None:
|
||||
raise ServiceNotFoundException(
|
||||
"The service you are searching for does not exist."
|
||||
)
|
||||
|
||||
self._session.delete(service_entity)
|
||||
self._session.commit()
|
||||
from fastapi import Depends
|
||||
|
||||
from ..database import db_session
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import func, select, and_, func, or_, exists, or_
|
||||
|
||||
from backend.models.service_model import Service
|
||||
from backend.models.user_model import User
|
||||
from backend.entities.service_entity import ServiceEntity
|
||||
from backend.models.enum_for_models import ProgramTypeEnum, UserTypeEnum
|
||||
from backend.services.exceptions import (
|
||||
ServiceNotFoundException,
|
||||
ProgramNotAssignedException,
|
||||
)
|
||||
|
||||
|
||||
class ServiceService:
|
||||
|
||||
def __init__(self, session: Session = Depends(db_session)):
|
||||
self._session = session
|
||||
|
||||
def get_service_by_user(self, subject: User) -> list[Service]:
|
||||
"""Resource method getting all of the resources that a user has access to based on role"""
|
||||
if subject.role != UserTypeEnum.VOLUNTEER:
|
||||
query = select(ServiceEntity)
|
||||
entities = self._session.scalars(query).all()
|
||||
return [service.to_model() for service in entities]
|
||||
else:
|
||||
programs = subject.program
|
||||
resources = []
|
||||
for program in programs:
|
||||
entities = self._session.query(ServiceEntity).where(ServiceEntity.program == program).all()
|
||||
for entity in entities:
|
||||
resources.append(entity.to_model())
|
||||
return [service for service in resources]
|
||||
|
||||
def get_service_by_program(self, program: ProgramTypeEnum) -> list[Service]:
|
||||
"""Service method getting services belonging to a particular program."""
|
||||
query = select(ServiceEntity).filter(ServiceEntity.program == program)
|
||||
entities = self._session.scalars(query)
|
||||
|
||||
return [entity.to_model() for entity in entities]
|
||||
|
||||
def get_service_by_id(self, id: int) -> Service:
|
||||
"""Service method getting services by id."""
|
||||
query = select(ServiceEntity).filter(ServiceEntity.id == id)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise ServiceNotFoundException(f"Service with id: {id} does not exist")
|
||||
|
||||
return entity.to_model()
|
||||
|
||||
def get_service_by_name(self, name: str) -> Service:
|
||||
"""Service method getting services by id."""
|
||||
query = select(ServiceEntity).filter(ServiceEntity.name == name)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise ServiceNotFoundException(f"Service with name: {name} does not exist")
|
||||
|
||||
return entity.to_model()
|
||||
|
||||
def create(self, subject: User, service: Service) -> Service:
|
||||
"""Creates/adds a service to the table."""
|
||||
if subject.role != UserTypeEnum.ADMIN:
|
||||
raise ProgramNotAssignedException(
|
||||
f"User is not {UserTypeEnum.ADMIN}, cannot create service"
|
||||
)
|
||||
|
||||
service_entity = ServiceEntity.from_model(service)
|
||||
self._session.add(service_entity)
|
||||
self._session.commit()
|
||||
return service_entity.to_model()
|
||||
|
||||
def update(self, subject: User, service: Service) -> Service:
|
||||
"""Updates a service if in the table."""
|
||||
if subject.role != UserTypeEnum.ADMIN:
|
||||
raise ProgramNotAssignedException(
|
||||
f"User is not {UserTypeEnum.ADMIN}, cannot update service"
|
||||
)
|
||||
|
||||
query = select(ServiceEntity).where(ServiceEntity.id == service.id)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise ServiceNotFoundException(
|
||||
"The service you are searching for does not exist."
|
||||
)
|
||||
|
||||
entity.name = service.name
|
||||
entity.status = service.status
|
||||
entity.summary = service.summary
|
||||
entity.requirements = service.requirements
|
||||
entity.program = service.program
|
||||
self._session.commit()
|
||||
|
||||
return entity.to_model()
|
||||
|
||||
def delete(self, subject: User, service: Service) -> None:
|
||||
"""Deletes a service from the table."""
|
||||
if subject.role != UserTypeEnum.ADMIN:
|
||||
raise ProgramNotAssignedException(f"User is not {UserTypeEnum.ADMIN}")
|
||||
|
||||
query = select(ServiceEntity).where(ServiceEntity.id == service.id)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise ServiceNotFoundException(
|
||||
"The service you are searching for does not exist."
|
||||
)
|
||||
|
||||
self._session.delete(entity)
|
||||
self._session.commit()
|
||||
|
|
|
|||
|
|
@ -1,20 +1,52 @@
|
|||
from fastapi import Depends
|
||||
from ..database import db_session
|
||||
from sqlalchemy.orm import Session
|
||||
from ..models.tag_model import Tag
|
||||
from ..entities.tag_entity import TagEntity
|
||||
from sqlalchemy import select
|
||||
|
||||
|
||||
class TagService:
|
||||
|
||||
def __init__(self, session: Session = Depends(db_session)):
|
||||
self._session = session
|
||||
|
||||
def all(self) -> list[Tag]:
|
||||
"""Returns a list of all Tags"""
|
||||
|
||||
query = select(TagEntity)
|
||||
entities = self._session.scalars(query).all()
|
||||
|
||||
return [entity.to_model() for entity in entities]
|
||||
from fastapi import Depends
|
||||
|
||||
from backend.models.enum_for_models import UserTypeEnum
|
||||
from backend.models.user_model import User
|
||||
from backend.services.exceptions import TagNotFoundException
|
||||
from ..database import db_session
|
||||
from sqlalchemy.orm import Session
|
||||
from ..models.tag_model import Tag
|
||||
from ..entities.tag_entity import TagEntity
|
||||
from sqlalchemy import select
|
||||
|
||||
# Add in checks for user permission?
|
||||
class TagService:
|
||||
|
||||
def __init__(self, session: Session = Depends(db_session)):
|
||||
self._session = session
|
||||
|
||||
def get_all(self) -> list[Tag]:
|
||||
"""Returns a list of all Tags"""
|
||||
query = select(TagEntity)
|
||||
entities = self._session.scalars(query).all()
|
||||
|
||||
return [entity.to_model() for entity in entities]
|
||||
|
||||
def create(self, subject: User, tag: Tag) -> Tag:
|
||||
entity = TagEntity.from_model(tag)
|
||||
self._session.add(entity)
|
||||
self._session.commit()
|
||||
return entity.to_model()
|
||||
|
||||
def update(self, subject: User, tag: Tag) -> Tag:
|
||||
query = select(TagEntity).where(TagEntity.id == tag.id)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise TagNotFoundException(f"Tag with id {tag.id} does not exist")
|
||||
|
||||
entity.content = tag.content
|
||||
self._session.commit()
|
||||
return entity.to_model()
|
||||
|
||||
|
||||
def delete(self, subject: User, tag: Tag) -> None:
|
||||
query = select(TagEntity).where(TagEntity.id == tag.id)
|
||||
entity = self._session.scalars(query).one_or_none()
|
||||
|
||||
if entity is None:
|
||||
raise TagNotFoundException(f"Tag with id {tag.id} does not exist")
|
||||
|
||||
self._session.delete(entity)
|
||||
self._session.commit()
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user