rushilwiz/compass
1
0
Fork 0
mirror of https://github.com/cssgunc/compass.git synced 2025-04-20 10:30:16 -04:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge 3814d6d290 into cb54c9829d

Browse Source
This commit is contained in:
Aidan Kim 2024-10-29 23:30:02 +00:00 committed by GitHub
commit e85f90c955
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 724 additions and 647 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
backend/api/resource.py
View File

@ -1,26 +1,56 @@
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends
from ..services import ResourceService, UserService
from ..models.resource_model import Resource from backend.models.user_model import User
from ..services import ResourceService, UserService
from typing import List from ..models.resource_model import Resource
api = APIRouter(prefix="/api/resource") from typing import List
openapi_tags = { api = APIRouter(prefix="/api/resource")
"name": "Resource",
"description": "Resource search and related operations.", 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 # TODO: Add security using HTTP Bearer Tokens
@api.get("", response_model=List[Resource], tags=["Resource"]) # TODO: Enable authorization by passing user uuid to API
def get_all( # TODO: Create custom exceptions
user_id: str, @api.post("", response_model=Resource, tags=["Resource"])
resource_svc: ResourceService = Depends(), def create(
user_svc: UserService = Depends(), uuid: str, resource: Resource, user_svc: UserService = Depends(), resource_svc: ResourceService = Depends()
): ):
subject = user_svc.get_user_by_uuid(user_id) subject = user_svc.get_user_by_uuid(uuid)
return resource_svc.create(subject, resource)
return resource_svc.get_resource_by_user(subject)
@api.get("", response_model=List[Resource], tags=["Resource"])
def get_all(
uuid: str, user_svc: UserService = Depends(), resource_svc: ResourceService = Depends()
):
subject = user_svc.get_user_by_uuid(uuid)
return resource_svc.get_resource_by_user(subject)
@api.get("/{name}", response_model=Resource, tags=["Resource"])
def get_by_name(
name:str, uuid:str, user_svc: UserService = Depends(), resource_svc: ResourceService = Depends()
):
subject = user_svc.get_user_by_uuid(uuid)
return resource_svc.get_resource_by_name(name, subject)
@api.put("", response_model=Resource, tags=["Resource"])
def update(
uuid: str, resource: Resource, user_svc: UserService = Depends(), resource_svc: ResourceService = Depends()
):
subject = user_svc.get_user_by_uuid(uuid)
return resource_svc.update(subject, resource)
@api.delete("", response_model=None, tags=["Resource"])
def delete(
uuid: str, resource: Resource, user_svc: UserService = Depends(), resource_svc: ResourceService = Depends()
):
subject = user_svc.get_user_by_uuid(uuid)
resource_svc.delete(subject, resource)

80
backend/api/service.py
View File

@ -1,26 +1,54 @@
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends
from ..services import ServiceService, UserService
from ..models.service_model import Service from backend.models.user_model import User
from ..services import ServiceService, UserService
from typing import List from ..models.service_model import Service
api = APIRouter(prefix="/api/service") from typing import List
openapi_tags = { api = APIRouter(prefix="/api/service")
"name": "Service",
"description": "Service search and related operations.", 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 # TODO: Add security using HTTP Bearer Tokens
@api.get("", response_model=List[Service], tags=["Service"]) # TODO: Enable authorization by passing user uuid to API
def get_all( # TODO: Create custom exceptions
user_id: str, @api.post("", response_model=Service, tags=["Service"])
service_svc: ServiceService = Depends(), def create(
user_svc: UserService = Depends(), uuid: str, service: Service, user_svc: UserService = Depends(), service_svc: ServiceService = Depends()
): ):
subject = user_svc.get_user_by_uuid(user_id) subject = user_svc.get_user_by_uuid(uuid)
return service_svc.create(subject, service)
return service_svc.get_service_by_user(subject)
@api.get("", response_model=List[Service], tags=["Service"])
def get_all(
uuid: str, user_svc: UserService = Depends(), service_svc: ServiceService = Depends()
):
subject = user_svc.get_user_by_uuid(uuid)
return service_svc.get_service_by_user(subject)
@api.get("/{name}", response_model=Service, tags=["Service"])
def get_by_name(
name: str, uuid: str, user_svc: UserService = Depends(), service_svc: ServiceService = Depends()
):
subject = user_svc.get_user_by_uuid(uuid)
return service_svc.get_service_by_name(name, subject)
@api.put("", response_model=Service, tags=["Service"])
def update(
uuid: str, service: Service, user_svc: UserService = Depends(), service_svc: ServiceService = Depends()
):
subject = user_svc.get_user_by_uuid(uuid)
return service_svc.update(subject, service)
@api.delete("", response_model=None, tags=["Service"])
def delete(
uuid: str, service: Service, user_svc: UserService = Depends(), service_svc: ServiceService = Depends()
):
subject = user_svc.get_user_by_uuid(uuid)
service_svc.delete(subject, service)

51
backend/api/tag.py Normal file
View 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=Depends()
):
return tag_service.create(subject, tag)
@api.get("", response_model=List[Tag], tags=["Tag"])
def get_all(
subject: User,
tag_svc: TagService=Depends()
):
return tag_svc.get_all()
@api.put("", response_model=Tag, tags=["Tag"])
def update(
subject: User,
tag: Tag,
tag_svc: TagService=Depends()
):
return tag_svc.delete(subject, tag)
@api.delete("", response_model=None, tags=["Tag"])
def delete(
subject: User,
tag: Tag,
tag_svc: TagService=Depends()
):
tag_svc.delete(subject, tag)

134
backend/entities/resource_entity.py
View File

@ -1,67 +1,67 @@
""" Defines the table for storing resources """ """ Defines the table for storing resources """
# Import our mapped SQL types from SQLAlchemy # Import our mapped SQL types from SQLAlchemy
from sqlalchemy import Integer, String, DateTime, Enum from sqlalchemy import Integer, String, DateTime, Enum
# Import mapping capabilities from the SQLAlchemy ORM # Import mapping capabilities from the SQLAlchemy ORM
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
# Import the EntityBase that we are extending # Import the EntityBase that we are extending
from .entity_base import EntityBase from .entity_base import EntityBase
# Import datetime for created_at type # Import datetime for created_at type
from datetime import datetime from datetime import datetime
# Import self for to model # Import self for to model
from typing import Self from typing import Self
from backend.entities.program_enum import Program_Enum from backend.entities.program_enum import Program_Enum
from ..models.resource_model import Resource from ..models.resource_model import Resource
class ResourceEntity(EntityBase): class ResourceEntity(EntityBase):
# set table name # set table name
__tablename__ = "resource" __tablename__ = "resource"
# set fields # set fields
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
name: Mapped[str] = mapped_column(String(64), nullable=False) name: Mapped[str] = mapped_column(String(64), nullable=False)
summary: Mapped[str] = mapped_column(String(100), nullable=False) summary: Mapped[str] = mapped_column(String(100), nullable=False)
link: Mapped[str] = mapped_column(String, nullable=False) link: Mapped[str] = mapped_column(String, nullable=False)
program: Mapped[Program_Enum] = mapped_column(Enum(Program_Enum), nullable=False) program: Mapped[Program_Enum] = mapped_column(Enum(Program_Enum), nullable=False)
# relationships # relationships
resourceTags: Mapped[list["ResourceTagEntity"]] = relationship( resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(
back_populates="resource", cascade="all,delete" back_populates="resource", cascade="all,delete"
) )
@classmethod @classmethod
def from_model(cls, model: Resource) -> Self: def from_model(cls, model: Resource) -> Self:
""" """
Create a UserEntity from a User model. Create a UserEntity from a User model.
Args: Args:
model (User): The model to create the entity from. model (User): The model to create the entity from.
Returns: Returns:
Self: The entity (not yet persisted). Self: The entity (not yet persisted).
""" """
return cls( return cls(
id=model.id, id=model.id,
created_at=model.created_at, created_at=model.created_at,
name=model.name, name=model.name,
summary=model.summary, summary=model.summary,
link=model.link, link=model.link,
program=model.program, program=model.program,
) )
def to_model(self) -> Resource: def to_model(self) -> Resource:
return Resource( return Resource(
id=self.id, id=self.id,
created_at=self.created_at, created_at=self.created_at,
name=self.name, name=self.name,
summary=self.summary, summary=self.summary,
link=self.link, link=self.link,
program=self.program, program=self.program,
) )

92
backend/entities/resource_tag_entity.py
View File

@ -1,46 +1,46 @@
""" Defines the table for resource tags """ """ Defines the table for resource tags """
# Import our mapped SQL types from SQLAlchemy # Import our mapped SQL types from SQLAlchemy
from sqlalchemy import ForeignKey, Integer, String, DateTime from sqlalchemy import ForeignKey, Integer, String, DateTime
# Import mapping capabilities from the SQLAlchemy ORM # Import mapping capabilities from the SQLAlchemy ORM
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
# Import the EntityBase that we are extending # Import the EntityBase that we are extending
from .entity_base import EntityBase from .entity_base import EntityBase
# Import datetime for created_at type # Import datetime for created_at type
from datetime import datetime from datetime import datetime
# Import self for to model # Import self for to model
from typing import Self from typing import Self
class ResourceTagEntity(EntityBase): class ResourceTagEntity(EntityBase):
# set table name to user in the database # set table name to user in the database
__tablename__ = "resource_tag" __tablename__ = "resource_tag"
# set fields or 'columns' for the user table # set fields or 'columns' for the user table
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
resourceId: Mapped[int] = mapped_column(ForeignKey("resource.id")) resourceId: Mapped[int] = mapped_column(ForeignKey("resource.id"))
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id")) tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
# relationships # relationships
resource: Mapped["ResourceEntity"] = relationship(back_populates="resourceTags") resource: Mapped["ResourceEntity"] = relationship(back_populates="resourceTags")
tag: Mapped["TagEntity"] = relationship(back_populates="resourceTags") tag: Mapped["TagEntity"] = relationship(back_populates="resourceTags")
# @classmethod # @classmethod
# def from_model (cls, model: resource_tag_model) -> Self: # def from_model (cls, model: resource_tag_model) -> Self:
# return cls ( # return cls (
# id = model.id, # id = model.id,
# resourceId = model.resourceId, # resourceId = model.resourceId,
# tagId = model.tagId, # tagId = model.tagId,
# ) # )
# def to_model (self) -> resource_tag_model: # def to_model (self) -> resource_tag_model:
# return user_model( # return user_model(
# id = self.id, # id = self.id,
# resourceId = self.resourceId, # resourceId = self.resourceId,
# tagId = self.tagId, # tagId = self.tagId,
# ) # )

92
backend/entities/service_entity.py
View File

@ -1,47 +1,47 @@
""" Defines the table for storing services """ """ Defines the table for storing services """
# Import our mapped SQL types from SQLAlchemy # Import our mapped SQL types from SQLAlchemy
from sqlalchemy import Integer, String, DateTime, ARRAY from sqlalchemy import Integer, String, DateTime, ARRAY
# Import mapping capabilities from the SQLAlchemy ORM # Import mapping capabilities from the SQLAlchemy ORM
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
# Import the EntityBase that we are extending # Import the EntityBase that we are extending
from .entity_base import EntityBase from .entity_base import EntityBase
# Import datetime for created_at type # Import datetime for created_at type
from datetime import datetime from datetime import datetime
# Import enums for Program # Import enums for Program
import enum import enum
from sqlalchemy import Enum from sqlalchemy import Enum
from backend.models.service_model import Service from backend.models.service_model import Service
from typing import Self from typing import Self
from backend.models.enum_for_models import ProgramTypeEnum from backend.models.enum_for_models import ProgramTypeEnum
class ServiceEntity(EntityBase): class ServiceEntity(EntityBase):
# set table name # set table name
__tablename__ = "service" __tablename__ = "service"
# set fields # set fields
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
name: Mapped[str] = mapped_column(String(32), nullable=False) name: Mapped[str] = mapped_column(String(32), nullable=False)
status: 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) summary: Mapped[str] = mapped_column(String(100), nullable=False)
requirements: Mapped[list[str]] = mapped_column(ARRAY(String)) requirements: Mapped[list[str]] = mapped_column(ARRAY(String))
program: Mapped[ProgramTypeEnum] = mapped_column(Enum(ProgramTypeEnum), nullable=False) program: Mapped[ProgramTypeEnum] = mapped_column(Enum(ProgramTypeEnum), nullable=False)
# relationships # relationships
serviceTags: Mapped[list["ServiceTagEntity"]] = relationship( serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(
back_populates="service", cascade="all,delete" back_populates="service", cascade="all,delete"
) )
def to_model(self) -> Service: 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) return Service(id=self.id, name=self.name, status=self.status, summary=self.summary, requirements=self.requirements, program=self.program)
@classmethod @classmethod
def from_model(cls, model:Service) -> Self: 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) return cls(id=model.id, name=model.name, status=model.status, summary=model.summary, requirements=model.requirements, program=model.program)

51
backend/entities/service_tag_entity.py
View File

@ -1,25 +1,26 @@
""" Defines the table for service tags """ """ Defines the table for service tags """
# Import our mapped SQL types from SQLAlchemy # Import our mapped SQL types from SQLAlchemy
from sqlalchemy import ForeignKey, Integer from sqlalchemy import ForeignKey, Integer
# Import mapping capabilities from the SQLAlchemy ORM # Import mapping capabilities from the SQLAlchemy ORM
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
# Import the EntityBase that we are extending # Import the EntityBase that we are extending
from .entity_base import EntityBase from .entity_base import EntityBase
class ServiceTagEntity(EntityBase): class ServiceTagEntity(EntityBase):
# set table name to user in the database # set table name to user in the database
__tablename__ = "service_tag" __tablename__ = "service_tag"
# set fields or 'columns' for the user table # set fields or 'columns' for the user table
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
serviceId: Mapped[int] = mapped_column(ForeignKey("service.id")) serviceId: Mapped[int] = mapped_column(ForeignKey("service.id"))
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id")) tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
# relationships # relationships
service: Mapped["ServiceEntity"] = relationship(back_populates="serviceTags") service: Mapped["ServiceEntity"] = relationship(back_populates="serviceTags")
tag: Mapped["TagEntity"] = relationship(back_populates="serviceTags") tag: Mapped["TagEntity"] = relationship(back_populates="serviceTags")

129
backend/entities/tag_entity.py
View File

@ -1,65 +1,64 @@
""" Defines the table for storing tags """ """ Defines the table for storing tags """
# Import our mapped SQL types from SQLAlchemy # Import our mapped SQL types from SQLAlchemy
from sqlalchemy import Integer, String, DateTime from sqlalchemy import Integer, String, DateTime
# Import mapping capabilities from the SQLAlchemy ORM # Import mapping capabilities from the SQLAlchemy ORM
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
# Import the EntityBase that we are extending # Import the EntityBase that we are extending
from .entity_base import EntityBase from .entity_base import EntityBase
# Import datetime for created_at type # Import datetime for created_at type
from datetime import datetime from datetime import datetime
from ..models.tag_model import Tag from ..models.tag_model import Tag
from typing import Self from typing import Self
class TagEntity(EntityBase): class TagEntity(EntityBase):
#set table name #set table name
__tablename__ = "tag" __tablename__ = "tag"
#set fields #set fields
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
content: Mapped[str] = mapped_column(String(100), nullable=False) content: Mapped[str] = mapped_column(String(100), nullable=False)
#relationships #relationships
resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(back_populates="tag", cascade="all,delete") resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(back_populates="tag", cascade="all,delete")
serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(back_populates="tag", cascade="all,delete") serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(back_populates="tag", cascade="all,delete")
@classmethod @classmethod
def from_model(cls, model: Tag) -> Self: def from_model(cls, model: Tag) -> Self:
""" """
Create a user entity from model Create a user entity from model
Args: model (User): the model to create the entity from Args: model (User): the model to create the entity from
Returns: Returns:
self: The entity self: The entity
""" """
return cls( return cls(
id=model.id, id=model.id,
content=model.id, created_at=model.created_at,
) content=model.id,
)
def to_model(self) -> Tag:
""" def to_model(self) -> Tag:
Create a user model from entity """
Create a user model from entity
Returns:
User: A User model for API usage Returns:
""" User: A User model for API usage
"""
return Tag(
id=self.id, return Tag(
content=self.content, id=self.id,
) create_at=self.created_at,
content=self.content,
)

70
backend/main.py
View File

@ -1,34 +1,36 @@
from fastapi import FastAPI, Request from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
from fastapi.middleware.gzip import GZipMiddleware from fastapi.middleware.gzip import GZipMiddleware
from .api import user, health, service, resource
from .api import user, health, service, resource, tag
description = """
Welcome to the **COMPASS** RESTful Application Programming Interface. description = """
""" Welcome to the **COMPASS** RESTful Application Programming Interface.
"""
app = FastAPI(
title="Compass API", app = FastAPI(
version="0.0.1", title="Compass API",
description=description, version="0.0.1",
openapi_tags=[ description=description,
user.openapi_tags, openapi_tags=[
health.openapi_tags, user.openapi_tags,
service.openapi_tags, health.openapi_tags,
resource.openapi_tags, service.openapi_tags,
], resource.openapi_tags,
) tag.openapi_tags
],
app.add_middleware(GZipMiddleware) )
feature_apis = [user, health, service, resource] app.add_middleware(GZipMiddleware)
for feature_api in feature_apis: feature_apis = [user, health, service, resource, tag]
app.include_router(feature_api.api)
for feature_api in feature_apis:
app.include_router(feature_api.api)
# Add application-wide exception handling middleware for commonly encountered API Exceptions
@app.exception_handler(Exception)
def permission_exception_handler(request: Request, e: Exception): # Add application-wide exception handling middleware for commonly encountered API Exceptions
return JSONResponse(status_code=403, content={"message": str(e)}) @app.exception_handler(Exception)
def permission_exception_handler(request: Request, e: Exception):
return JSONResponse(status_code=403, content={"message": str(e)})

2
backend/services/exceptions.py
View File

@ -20,6 +20,8 @@ class UserPermissionException(Exception):
class ServiceNotFoundException(Exception): class ServiceNotFoundException(Exception):
"""Exception for when the service being requested is not in the table.""" """Exception for when the service being requested is not in the table."""
class TagNotFoundException(Exception):
"""Exception for when the tag being requested is not in the table."""
class ProgramNotAssignedException(Exception): class ProgramNotAssignedException(Exception):
"""Exception for when the user does not have correct access for requested services.""" """Exception for when the user does not have correct access for requested services."""

288
backend/services/resource.py
View File

@ -1,165 +1,123 @@
from fastapi import Depends from fastapi import Depends
from ..database import db_session from ..database import db_session
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from sqlalchemy import select from sqlalchemy import and_, select
from ..models.resource_model import Resource from ..models.resource_model import Resource
from ..entities.resource_entity import ResourceEntity from ..entities.resource_entity import ResourceEntity
from ..models.user_model import User, UserTypeEnum from ..models.user_model import User, UserTypeEnum
from .exceptions import ResourceNotFoundException from .exceptions import ProgramNotAssignedException, ResourceNotFoundException
class ResourceService: class ResourceService:
def __init__(self, session: Session = Depends(db_session)): def __init__(self, session: Session = Depends(db_session)):
self._session = session self._session = session
def get_resource_by_user(self, subject: User): 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""" """Resource method getting all of the resources that a user has access to based on role"""
if subject.role != UserTypeEnum.VOLUNTEER: if subject.role != UserTypeEnum.VOLUNTEER:
query = select(ResourceEntity) query = select(ResourceEntity)
entities = self._session.scalars(query).all() entities = self._session.scalars(query).all()
return [resource.to_model() for resource in entities]
return [resource.to_model() for resource in entities] else:
else: programs = subject.program
programs = subject.program resources = []
resources = [] for program in programs:
for program in programs: entities = (
query = select(ResourceEntity).filter(ResourceEntity.program == program) self._session.query(ResourceEntity)
entities = self._session.scalars(query).all() .where(ResourceEntity.program == program)
for entity in entities: .all()
resources.append(entity) )
for entity in entities:
return [resource.to_model() for resource in resources] resources.append(entity.to_model())
return [resource for resource in resources]
def create(self, user: User, resource: Resource) -> Resource:
""" def get_resource_by_name(self, name: str, subject: User) -> Resource:
Creates a resource based on the input object and adds it to the table if the user has the right permissions. """Get a resource by name."""
query = select(ResourceEntity).where(
Parameters: and_(
user: a valid User model representing the currently logged in User ResourceEntity.name == name, ResourceEntity.program.in_(subject.program)
resource: Resource object to add to table )
)
Returns: entity = self._session.scalars(query).one_or_none()
Resource: Object added to table if entity is None:
""" raise ResourceNotFoundException(
if resource.role != user.role or resource.group != user.group: f"Resource with name: {name} does not exist or program has not been assigned."
raise PermissionError( )
"User does not have permission to add resources in this role or group." return entity.to_model()
)
def create(self, subject: User, resource: Resource) -> Resource:
resource_entity = ResourceEntity.from_model(resource) """
self._session.add(resource_entity) Creates a resource based on the input object and adds it to the table if the user has the right permissions.
self._session.commit()
Parameters:
return resource_entity.to_model() user: a valid User model representing the currently logged in User
resource: Resource object to add to table
def get_by_id(self, user: User, id: int) -> Resource:
""" Returns:
Gets a resource based on the resource id that the user has access to Resource: Object added to table
"""
Parameters: if subject.role != UserTypeEnum.ADMIN:
user: a valid User model representing the currently logged in User raise ProgramNotAssignedException(
id: int, the id of the resource f"User is not {UserTypeEnum.ADMIN}, cannot update service"
)
Returns: resource_entity = ResourceEntity.from_model(resource)
Resource self._session.add(resource_entity)
self._session.commit()
Raises: return resource_entity.to_model()
ResourceNotFoundException: If no resource is found with id
""" def update(self, subject: User, resource: Resource) -> Resource:
resource = ( """
self._session.query(ResourceEntity) Update the resource if the user has access
.filter(
ResourceEntity.id == id, Parameters:
ResourceEntity.role == user.role, user: a valid User model representing the currently logged in User
ResourceEntity.group == user.group, resource (ResourceEntity): Resource to update
)
.one_or_none() Returns:
) Resource: Updated resource object
if resource is None: Raises:
raise ResourceNotFoundException(f"No resource found with id: {id}") ResourceNotFoundException: If no resource is found with the corresponding ID
"""
return resource.to_model() if subject.role != UserTypeEnum.ADMIN:
raise ProgramNotAssignedException(
def update(self, user: User, resource: ResourceEntity) -> Resource: f"User is not {UserTypeEnum.ADMIN}, cannot update service"
""" )
Update the resource if the user has access query = select(ResourceEntity).where(ResourceEntity.id == resource.id)
entity = self._session.scalars(query).one_or_none()
Parameters: if entity is None:
user: a valid User model representing the currently logged in User raise ResourceNotFoundException(
resource (ResourceEntity): Resource to update f"No resource found with matching id: {resource.id}"
)
Returns: entity.name = resource.name
Resource: Updated resource object entity.summary = resource.summary
entity.link = resource.link
Raises: entity.program = resource.program
ResourceNotFoundException: If no resource is found with the corresponding ID self._session.commit()
""" return entity.to_model()
if resource.role != user.role or resource.group != user.group:
raise PermissionError( def delete(self, subject: User, resource: Resource) -> None:
"User does not have permission to update this resource." """
) Delete resource based on id that the user has access to
obj = self._session.get(ResourceEntity, resource.id) if resource.id else None Parameters:
user: a valid User model representing the currently logged in User
if obj is None: id: int, a unique resource id
raise ResourceNotFoundException(
f"No resource found with matching id: {resource.id}" Raises:
) ResourceNotFoundException: If no resource is found with the corresponding id
"""
obj.update_from_model(resource) # Assuming an update method exists if subject.role != UserTypeEnum.ADMIN:
self._session.commit() raise ProgramNotAssignedException(
f"User is not {UserTypeEnum.ADMIN}, cannot update service"
return obj.to_model() )
query = select(ResourceEntity).where(ResourceEntity.id == resource.id)
def delete(self, user: User, id: int) -> None: entity = self._session.scalars(query).one_or_none()
""" if entity is None:
Delete resource based on id that the user has access to raise ResourceNotFoundException(
f"No resource found with matching id: {resource.id}"
Parameters: )
user: a valid User model representing the currently logged in User self._session.delete(entity)
id: int, a unique resource id self._session.commit()
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]

228
backend/services/service.py
View File

@ -1,127 +1,101 @@
from fastapi import Depends from fastapi import Depends
from ..database import db_session from ..database import db_session
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from sqlalchemy import func, select, and_, func, or_, exists, or_ from sqlalchemy import func, select, and_, func, or_, exists, or_
from backend.models.service_model import Service from backend.models.service_model import Service
from backend.models.user_model import User from backend.models.user_model import User
from backend.entities.service_entity import ServiceEntity from backend.entities.service_entity import ServiceEntity
from backend.models.enum_for_models import ProgramTypeEnum, UserTypeEnum from backend.models.enum_for_models import ProgramTypeEnum, UserTypeEnum
from backend.services.exceptions import ( from backend.services.exceptions import (
ServiceNotFoundException, ServiceNotFoundException,
ProgramNotAssignedException, ProgramNotAssignedException,
) )
class ServiceService: class ServiceService:
def __init__(self, session: Session = Depends(db_session)): def __init__(self, session: Session = Depends(db_session)):
self._session = session self._session = session
def get_service_by_program(self, program: ProgramTypeEnum) -> list[Service]: def get_service_by_user(self, subject: User) -> list[Service]:
"""Service method getting services belonging to a particular program.""" """Resource method getting all of the resources that a user has access to based on role"""
query = select(ServiceEntity).filter(ServiceEntity.program == program) if subject.role != UserTypeEnum.VOLUNTEER:
entities = self._session.scalars(query) query = select(ServiceEntity)
entities = self._session.scalars(query).all()
return [entity.to_model() for entity in entities] return [service.to_model() for service in entities]
else:
def get_service_by_id(self, id: int) -> Service: programs = subject.program
"""Service method getting services by id.""" services = []
query = select(ServiceEntity).filter(ServiceEntity.id == id) for program in programs:
entity = self._session.scalars(query).one_or_none() entities = self._session.query(ServiceEntity).where(ServiceEntity.program == program).all()
for entity in entities:
if entity is None: services.append(entity.to_model())
raise ServiceNotFoundException(f"Service with id: {id} does not exist") return [service for service in services]
return entity.to_model() def get_service_by_name(self, name: str, subject: User) -> Service:
"""Service method getting services by id."""
def get_service_by_name(self, name: str) -> Service: query = select(ServiceEntity).where(
"""Service method getting services by id.""" and_(
query = select(ServiceEntity).filter(ServiceEntity.name == name) ServiceEntity.name == name, ServiceEntity.program.in_(subject.program)
entity = self._session.scalars(query).one_or_none() )
)
if entity is None: entity = self._session.scalars(query).one_or_none()
raise ServiceNotFoundException(f"Service with name: {name} does not exist")
if entity is None:
return entity.to_model() raise ServiceNotFoundException(f"Service with name: {name} does not exist or program has not been assigned")
def get_service_by_user(self, subject: User): return entity.to_model()
"""Service method getting all of the services that a user has access to based on role"""
if subject.role != UserTypeEnum.VOLUNTEER: def create(self, subject: User, service: Service) -> Service:
query = select(ServiceEntity) """Creates/adds a service to the table."""
entities = self._session.scalars(query).all() if subject.role != UserTypeEnum.ADMIN:
raise ProgramNotAssignedException(
return [service.to_model() for service in entities] f"User is not {UserTypeEnum.ADMIN}, cannot create service"
else: )
programs = subject.program
services = [] service_entity = ServiceEntity.from_model(service)
for program in programs: self._session.add(service_entity)
query = select(ServiceEntity).filter(ServiceEntity.program == program) self._session.commit()
entities = self._session.scalars(query).all() return service_entity.to_model()
for entity in entities:
services.append(entity) def update(self, subject: User, service: Service) -> Service:
"""Updates a service if in the table."""
return [service.to_model() for service in services] if subject.role != UserTypeEnum.ADMIN:
raise ProgramNotAssignedException(
def get_all(self, subject: User) -> list[Service]: f"User is not {UserTypeEnum.ADMIN}, cannot update service"
"""Service method retrieving all of the services in the table.""" )
if subject.role == UserTypeEnum.VOLUNTEER:
raise ProgramNotAssignedException( query = select(ServiceEntity).where(ServiceEntity.id == service.id)
f"User is not {UserTypeEnum.ADMIN} or {UserTypeEnum.VOLUNTEER}, cannot get all" entity = self._session.scalars(query).one_or_none()
)
if entity is None:
query = select(ServiceEntity) raise ServiceNotFoundException(
entities = self._session.scalars(query).all() "The service you are searching for does not exist."
)
return [service.to_model() for service in entities]
entity.name = service.name
def create(self, subject: User, service: Service) -> Service: entity.status = service.status
"""Creates/adds a service to the table.""" entity.summary = service.summary
if subject.role != UserTypeEnum.ADMIN: entity.requirements = service.requirements
raise ProgramNotAssignedException( entity.program = service.program
f"User is not {UserTypeEnum.ADMIN}, cannot create service" self._session.commit()
)
return entity.to_model()
service_entity = ServiceEntity.from_model(service)
self._session.add(service_entity) def delete(self, subject: User, service: Service) -> None:
self._session.commit() """Deletes a service from the table."""
return service_entity.to_model() if subject.role != UserTypeEnum.ADMIN:
raise ProgramNotAssignedException(f"User is not {UserTypeEnum.ADMIN}")
def update(self, subject: User, service: Service) -> Service:
"""Updates a service if in the table.""" query = select(ServiceEntity).where(ServiceEntity.id == service.id)
if subject.role != UserTypeEnum.ADMIN: entity = self._session.scalars(query).one_or_none()
raise ProgramNotAssignedException(
f"User is not {UserTypeEnum.ADMIN}, cannot update service" if entity is None:
) raise ServiceNotFoundException(
"The service you are searching for does not exist."
service_entity = self._session.get(ServiceEntity, service.id) )
if service_entity is None: self._session.delete(entity)
raise ServiceNotFoundException( self._session.commit()
"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()

72
backend/services/tag.py
View File

@ -1,20 +1,52 @@
from fastapi import Depends from fastapi import Depends
from ..database import db_session
from sqlalchemy.orm import Session from backend.models.enum_for_models import UserTypeEnum
from ..models.tag_model import Tag from backend.models.user_model import User
from ..entities.tag_entity import TagEntity from backend.services.exceptions import TagNotFoundException
from sqlalchemy import select from ..database import db_session
from sqlalchemy.orm import Session
from ..models.tag_model import Tag
class TagService: from ..entities.tag_entity import TagEntity
from sqlalchemy import select
def __init__(self, session: Session = Depends(db_session)):
self._session = session # Add in checks for user permission?
class TagService:
def all(self) -> list[Tag]:
"""Returns a list of all Tags""" def __init__(self, session: Session = Depends(db_session)):
self._session = session
query = select(TagEntity)
entities = self._session.scalars(query).all() def get_all(self) -> list[Tag]:
"""Returns a list of all Tags"""
return [entity.to_model() for entity in entities] 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()