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

Implement resource api methods

Browse Source
This commit is contained in:
pmoharana-cmd 2024-04-24 20:18:09 -04:00
parent 4e090e5bd5
commit ba15bf7519
8 changed files with 196 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
backend/api/resource.py Normal file
View File

@ -0,0 +1,26 @@
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)

54
backend/entities/resource_entity.py
View File

@ -15,6 +15,7 @@ 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):
@ -34,34 +35,33 @@ class ResourceEntity(EntityBase):
back_populates="resource", cascade="all,delete"
)
#
# @classmethod
# def from_model(cls, model: user_model) -> Self:
# """
# Create a UserEntity from a User model.
@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.
Args:
model (User): The model to create the entity from.
# Returns:
# Self: The entity (not yet persisted).
# """
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,
# )
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) -> user_model:
# return user_model (
# id = self.id,
# created_at = self.created_at,
# name = self.name,
# summary = self.summary,
# link = self.link,
# program = self.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,
)

11
backend/main.py
View File

@ -2,7 +2,7 @@ from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.middleware.gzip import GZipMiddleware
from .api import user, health, service
from .api import user, health, service, resource
description = """
Welcome to the **COMPASS** RESTful Application Programming Interface.
@ -12,12 +12,17 @@ app = FastAPI(
title="Compass API",
version="0.0.1",
description=description,
openapi_tags=[user.openapi_tags, health.openapi_tags, service.openapi_tags],
openapi_tags=[
user.openapi_tags,
health.openapi_tags,
service.openapi_tags,
resource.openapi_tags,
],
)
app.add_middleware(GZipMiddleware)
feature_apis = [user, health, service]
feature_apis = [user, health, service, resource]
for feature_api in feature_apis:
app.include_router(feature_api.api)

3
backend/models/resource_model.py
View File

@ -4,7 +4,6 @@ from typing import List
from datetime import datetime
from typing import Optional
from .enum_for_models import ProgramTypeEnum
from .resource_model import Resource
class Resource(BaseModel):
@ -12,5 +11,5 @@ class Resource(BaseModel):
name: str = Field(..., max_length=150, description="The name of the resource")
summary: str = Field(..., max_length=300, description="The summary of the resource")
link: str = Field(..., max_length=150, description="link to the resource")
programtype: ProgramTypeEnum
program: ProgramTypeEnum
created_at: Optional[datetime]

4
backend/script/reset_demo.py
View File

@ -6,7 +6,7 @@ from ..database import engine, _engine_str
from ..env import getenv
from .. import entities
from ..test.services import user_test_data, service_test_data
from ..test.services import user_test_data, service_test_data, resource_test_data
database = getenv("POSTGRES_DATABASE")
@ -23,5 +23,5 @@ entities.EntityBase.metadata.create_all(engine)
with Session(engine) as session:
user_test_data.insert_test_data(session)
service_test_data.insert_fake_data(session)
resource_test_data.insert_fake_data(session)
session.commit()

60
backend/services/resource.py
View File

@ -4,33 +4,33 @@ 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
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 all(self, user: User) -> list[Resource]:
"""
Retrieves all Resources that the user has access to based on their role and group.
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()
Parameters:
user: a valid User model representing the currently logged in User
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)
Returns:
list[Resource]: list of accessible `Resource` for the user
"""
# Filter resources based on user's role and group
query = select(ResourceEntity).where(
ResourceEntity.role == user.role,
ResourceEntity.group == user.group
)
entities = self._session.scalars(query).all()
return [entity.to_model() for entity in entities]
return [resource.to_model() for resource in resources]
def create(self, user: User, resource: Resource) -> Resource:
"""
@ -44,7 +44,9 @@ class ResourceService:
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.")
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)
@ -68,7 +70,11 @@ class ResourceService:
"""
resource = (
self._session.query(ResourceEntity)
.filter(ResourceEntity.id == id, ResourceEntity.role == user.role, ResourceEntity.group == user.group)
.filter(
ResourceEntity.id == id,
ResourceEntity.role == user.role,
ResourceEntity.group == user.group,
)
.one_or_none()
)
@ -92,12 +98,16 @@ class ResourceService:
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.")
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}")
raise ResourceNotFoundException(
f"No resource found with matching id: {resource.id}"
)
obj.update_from_model(resource) # Assuming an update method exists
self._session.commit()
@ -117,7 +127,11 @@ class ResourceService:
"""
resource = (
self._session.query(ResourceEntity)
.filter(ResourceEntity.id == id, ResourceEntity.role == user.role, ResourceEntity.group == user.group)
.filter(
ResourceEntity.id == id,
ResourceEntity.role == user.role,
ResourceEntity.group == user.group,
)
.one_or_none()
)
@ -144,7 +158,7 @@ class ResourceService:
query = select(ResourceEntity).where(
ResourceEntity.title.ilike(f"%{search_string}%"),
ResourceEntity.role == user.role,
ResourceEntity.group == user.group
ResourceEntity.group == user.group,
)
entities = self._session.scalars(query).all()

107
backend/test/services/resource_test.py
View File

@ -1,107 +0,0 @@
import pytest
from sqlalchemy.orm import Session
from sqlalchemy.exc import NoResultFound
from .resource_service import ResourceService
from .models.resource_model import Resource
from .entities.resource_entity import ResourceEntity
from .models.user_model import User
from .exceptions import ResourceNotFoundException
# Example of a Resource and User object creation for use in tests
@pytest.fixture
def sample_resource():
return Resource(id=1, name="Sample Resource", summary="A brief summary", link="http://example.com", programtype="TypeA")
@pytest.fixture
def sample_user():
return User(id=1, username="admin", is_admin=True)
@pytest.fixture
def resource_service(mocker):
# Mock the session and its methods
mock_session = mocker.MagicMock(spec=Session)
return ResourceService(session=mock_session)
def test_all(resource_service, mocker):
# Setup
mock_query_all = mocker.MagicMock(return_value=[ResourceEntity(id=1, name="Resource One"), ResourceEntity(id=2, name="Resource Two")])
mocker.patch.object(resource_service._session, 'scalars', return_value=mock_query_all)
# Execution
results = resource_service.all()
# Verification
assert len(results) == 2
assert results[0].id == 1
assert results[1].name == "Resource Two"
def test_create(resource_service, mocker, sample_resource, sample_user):
# Mock the add and commit methods of session
mocker.patch.object(resource_service._session, 'add')
mocker.patch.object(resource_service._session, 'commit')
# Execution
result = resource_service.create(sample_user, sample_resource)
# Verification
resource_service._session.add.assert_called_once()
resource_service._session.commit.assert_called_once()
assert result.id == sample_resource.id
assert result.name == sample_resource.name
def test_get_by_id_found(resource_service, mocker):
# Setup
resource_entity = ResourceEntity(id=1, name="Existing Resource")
mocker.patch.object(resource_service._session, 'query', return_value=mocker.MagicMock(one_or_none=mocker.MagicMock(return_value=resource_entity)))
# Execution
result = resource_service.get_by_id(1)
# Verification
assert result.id == 1
assert result.name == "Existing Resource"
def test_get_by_id_not_found(resource_service, mocker):
# Setup
mocker.patch.object(resource_service._session, 'query', return_value=mocker.MagicMock(one_or_none=mocker.MagicMock(return_value=None)))
# Execution & Verification
with pytest.raises(ResourceNotFoundException):
resource_service.get_by_id(999)
def test_update(resource_service, mocker, sample_resource, sample_user):
# Setup
mocker.patch.object(resource_service._session, 'get', return_value=sample_resource)
mocker.patch.object(resource_service._session, 'commit')
# Execution
result = resource_service.update(sample_user, sample_resource)
# Verification
assert result.id == sample_resource.id
resource_service._session.commit.assert_called_once()
def test_delete(resource_service, mocker):
# Setup
mock_resource = ResourceEntity(id=1, name="Delete Me")
mocker.patch.object(resource_service._session, 'query', return_value=mocker.MagicMock(one_or_none=mocker.MagicMock(return_value=mock_resource)))
mocker.patch.object(resource_service._session, 'delete')
mocker.patch.object(resource_service._session, 'commit')
# Execution
resource_service.delete(sample_user(), 1)
# Verification
resource_service._session.delete.assert_called_with(mock_resource)
resource_service._session.commit.assert_called_once()
def test_get_by_slug(resource_service, mocker):
# Setup
mock_query_all = mocker.MagicMock(return_value=[ResourceEntity(id=1, name="Resource One"), ResourceEntity(id=2, name="Resource Two")])
mocker.patch.object(resource_service._session, 'scalars', return_value=mock_query_all)
# Execution
results = resource_service.get_by_slug("Resource")
# Verification

95
backend/test/services/resource_test_data.py Normal file
View File

@ -0,0 +1,95 @@
from sqlalchemy.orm import Session
from datetime import datetime
from ...entities import ResourceEntity
from ...models.enum_for_models import ProgramTypeEnum
from ...models.resource_model import Resource
resource_1 = Resource(
id=1,
name="Resource 1",
summary="Helpful information for victims of domestic violence",
link="https://example.com/resource1",
program=ProgramTypeEnum.DOMESTIC,
created_at=datetime(2023, 6, 1, 10, 0, 0),
)
resource_2 = Resource(
id=2,
name="Resource 2",
summary="Legal assistance resources",
link="https://example.com/resource2",
program=ProgramTypeEnum.COMMUNITY,
created_at=datetime(2023, 6, 2, 12, 30, 0),
)
resource_3 = Resource(
id=3,
name="Resource 3",
summary="Financial aid resources",
link="https://example.com/resource3",
program=ProgramTypeEnum.ECONOMIC,
created_at=datetime(2023, 6, 3, 15, 45, 0),
)
resource_4 = Resource(
id=4,
name="Resource 4",
summary="Counseling and support groups",
link="https://example.com/resource4",
program=ProgramTypeEnum.DOMESTIC,
created_at=datetime(2023, 6, 4, 9, 15, 0),
)
resource_5 = Resource(
id=5,
name="Resource 5",
summary="Shelter and housing resources",
link="https://example.com/resource5",
program=ProgramTypeEnum.DOMESTIC,
created_at=datetime(2023, 6, 5, 11, 30, 0),
)
resources = [resource_1, resource_2, resource_3, resource_4, resource_5]
from sqlalchemy import text
from sqlalchemy.orm import Session, DeclarativeBase, InstrumentedAttribute
def reset_table_id_seq(
session: Session,
entity: type[DeclarativeBase],
entity_id_column: InstrumentedAttribute[int],
next_id: int,
) -> None:
"""Reset the ID sequence of an entity table.
Args:
session (Session) - A SQLAlchemy Session
entity (DeclarativeBase) - The SQLAlchemy Entity table to target
entity_id_column (MappedColumn) - The ID column (should be an int column)
next_id (int) - Where the next inserted, autogenerated ID should begin
Returns:
None"""
table = entity.__table__
id_column_name = entity_id_column.name
sql = text(f"ALTER SEQUENCe {table}_{id_column_name}_seq RESTART WITH {next_id}")
session.execute(sql)
def insert_fake_data(session: Session):
"""Inserts fake resource data into the test session."""
global resources
# Create entities for test resource data
entities = []
for resource in resources:
entity = ResourceEntity.from_model(resource)
session.add(entity)
entities.append(entity)
# Reset table IDs to prevent ID conflicts
reset_table_id_seq(session, ResourceEntity, ResourceEntity.id, len(resources) + 1)
# Commit all changes
session.commit()