mirror of
https://github.com/cssgunc/compass.git
synced 2025-04-21 18:59:49 -04:00
Compare commits
No commits in common. "e3dbe59fb905ff526fc1c21b45d95846cb76bc0b" and "68f07b8c5347fe43a4f8327f7d28e2e393bb100f" have entirely different histories.
e3dbe59fb9
...
68f07b8c53
|
@ -24,58 +24,3 @@ def get_all(
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
subject = user_svc.get_user_by_uuid(user_id)
|
||||||
|
|
||||||
return resource_svc.get_resource_by_user(subject)
|
return resource_svc.get_resource_by_user(subject)
|
||||||
|
|
||||||
|
|
||||||
@api.get("/{id}", response_model=Resource, tags=["Resource"])
|
|
||||||
def get_by_id(
|
|
||||||
user_id: str,
|
|
||||||
id: int,
|
|
||||||
resource_svc: ResourceService = Depends(),
|
|
||||||
user_svc: UserService = Depends(),
|
|
||||||
):
|
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
|
||||||
resource = resource_svc.get_resource_by_id(id)
|
|
||||||
return resource
|
|
||||||
|
|
||||||
|
|
||||||
@api.post("/", response_model=Resource, tags=["Resource"])
|
|
||||||
def create_service(
|
|
||||||
user_id: str,
|
|
||||||
resource: Resource,
|
|
||||||
resource_svc: ResourceService = Depends(),
|
|
||||||
user_svc: UserService = Depends(),
|
|
||||||
):
|
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
|
||||||
new_resource = resource_svc.create(subject, resource)
|
|
||||||
return new_resource
|
|
||||||
|
|
||||||
|
|
||||||
@api.put("/{resource_id}", response_model=Resource, tags=["Resource"])
|
|
||||||
def update_service(
|
|
||||||
resource_id: int,
|
|
||||||
user_id: str,
|
|
||||||
resource: Resource,
|
|
||||||
resource_svc: ResourceService = Depends(),
|
|
||||||
user_svc: UserService = Depends(),
|
|
||||||
):
|
|
||||||
resource.id = resource_id
|
|
||||||
|
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
|
||||||
updated_resource = resource_svc.update(subject, resource)
|
|
||||||
return updated_resource
|
|
||||||
|
|
||||||
|
|
||||||
@api.delete("/{resource_id}", response_model=Resource, tags=["Resource"])
|
|
||||||
def delete_service_tag_by_id(
|
|
||||||
resource_id: int,
|
|
||||||
tag_id: int,
|
|
||||||
user_id: str,
|
|
||||||
resource_svc: ResourceService = Depends(),
|
|
||||||
user_svc: UserService = Depends(),
|
|
||||||
):
|
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
|
||||||
resource = resource_svc.get_resource_by_id(resource_id)
|
|
||||||
tag = resource_svc._tag_service.get_tag_by_id(tag_id)
|
|
||||||
|
|
||||||
resource_svc.remove_tag(subject, resource, tag)
|
|
||||||
return resource_svc.get_resource_by_id(resource_id)
|
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
from fastapi import APIRouter, Depends, HTTPException
|
from fastapi import APIRouter, Depends
|
||||||
from ..services import ServiceService, UserService, TagService
|
from ..services import ServiceService, UserService
|
||||||
from ..models import Service, Tag
|
from ..models.service_model import Service
|
||||||
|
|
||||||
from typing import List
|
from typing import List
|
||||||
|
|
||||||
|
@ -24,58 +24,3 @@ def get_all(
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
subject = user_svc.get_user_by_uuid(user_id)
|
||||||
|
|
||||||
return service_svc.get_service_by_user(subject)
|
return service_svc.get_service_by_user(subject)
|
||||||
|
|
||||||
|
|
||||||
@api.get("/{id}", response_model=Service, tags=["Service"])
|
|
||||||
def get_by_id(
|
|
||||||
user_id: str,
|
|
||||||
id: int,
|
|
||||||
service_svc: ServiceService = Depends(),
|
|
||||||
user_svc: UserService = Depends(),
|
|
||||||
):
|
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
|
||||||
service = service_svc.get_service_by_id(id)
|
|
||||||
return service
|
|
||||||
|
|
||||||
|
|
||||||
@api.post("/", response_model=Service, tags=["Service"])
|
|
||||||
def create_service(
|
|
||||||
user_id: str,
|
|
||||||
service: Service,
|
|
||||||
service_svc: ServiceService = Depends(),
|
|
||||||
user_svc: UserService = Depends(),
|
|
||||||
):
|
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
|
||||||
new_service = service_svc.create(subject, service)
|
|
||||||
return new_service
|
|
||||||
|
|
||||||
|
|
||||||
@api.put("/{service_id}", response_model=Service, tags=["Service"])
|
|
||||||
def update_service(
|
|
||||||
service_id: int,
|
|
||||||
user_id: str,
|
|
||||||
service: Service,
|
|
||||||
service_svc: ServiceService = Depends(),
|
|
||||||
user_svc: UserService = Depends(),
|
|
||||||
):
|
|
||||||
service.id = service_id
|
|
||||||
|
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
|
||||||
updated_service = service_svc.update(subject, service)
|
|
||||||
return updated_service
|
|
||||||
|
|
||||||
|
|
||||||
@api.delete("/{service_id}", response_model=Service, tags=["Service"])
|
|
||||||
def delete_service_tag_by_id(
|
|
||||||
service_id: int,
|
|
||||||
tag_id: int,
|
|
||||||
user_id: str,
|
|
||||||
service_svc: ServiceService = Depends(),
|
|
||||||
user_svc: UserService = Depends(),
|
|
||||||
):
|
|
||||||
subject = user_svc.get_user_by_uuid(user_id)
|
|
||||||
service = service_svc.get_service_by_id(service_id)
|
|
||||||
tag = service_svc._tag_service.get_tag_by_id(tag_id)
|
|
||||||
|
|
||||||
service_svc.remove_tag(subject, service, tag)
|
|
||||||
return service_svc.get_service_by_id(service_id)
|
|
||||||
|
|
|
@ -31,8 +31,8 @@ class ResourceEntity(EntityBase):
|
||||||
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
|
||||||
resource_tags: Mapped[list["ResourceTagEntity"]] = relationship(
|
resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(
|
||||||
back_populates="resource", cascade="all,delete-orphan"
|
back_populates="resource", cascade="all,delete"
|
||||||
)
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
|
@ -64,5 +64,4 @@ class ResourceEntity(EntityBase):
|
||||||
summary=self.summary,
|
summary=self.summary,
|
||||||
link=self.link,
|
link=self.link,
|
||||||
program=self.program,
|
program=self.program,
|
||||||
tags=[tag_entity.tag.to_model() for tag_entity in self.resource_tags],
|
|
||||||
)
|
)
|
||||||
|
|
|
@ -1,7 +1,6 @@
|
||||||
""" 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 importlib.resources import Resource
|
|
||||||
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
|
||||||
|
@ -16,8 +15,6 @@ from datetime import datetime
|
||||||
# Import self for to model
|
# Import self for to model
|
||||||
from typing import Self
|
from typing import Self
|
||||||
|
|
||||||
from ..models import ResourceTag
|
|
||||||
|
|
||||||
|
|
||||||
class ResourceTagEntity(EntityBase):
|
class ResourceTagEntity(EntityBase):
|
||||||
|
|
||||||
|
@ -30,20 +27,20 @@ class ResourceTagEntity(EntityBase):
|
||||||
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
|
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
|
||||||
|
|
||||||
# relationships
|
# relationships
|
||||||
resource: Mapped["ResourceEntity"] = relationship(back_populates="resource_tags")
|
resource: Mapped["ResourceEntity"] = relationship(back_populates="resourceTags")
|
||||||
tag: Mapped["TagEntity"] = relationship(back_populates="resource")
|
tag: Mapped["TagEntity"] = relationship(back_populates="resourceTags")
|
||||||
|
|
||||||
def to_model(self) -> ResourceTag:
|
# @classmethod
|
||||||
return ResourceTag(
|
# def from_model (cls, model: resource_tag_model) -> Self:
|
||||||
id=self.id,
|
# return cls (
|
||||||
resourceId=self.resourceId,
|
# id = model.id,
|
||||||
tagId=self.tagId,
|
# resourceId = model.resourceId,
|
||||||
)
|
# tagId = model.tagId,
|
||||||
|
# )
|
||||||
|
|
||||||
@classmethod
|
# def to_model (self) -> resource_tag_model:
|
||||||
def from_model(cls, model: "ResourceTag") -> Self:
|
# return user_model(
|
||||||
return cls(
|
# id = self.id,
|
||||||
id=model.id,
|
# resourceId = self.resourceId,
|
||||||
resourceId=model.resourceId,
|
# tagId = self.tagId,
|
||||||
tagId=model.tagId,
|
# )
|
||||||
)
|
|
||||||
|
|
|
@ -20,7 +20,6 @@ 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
|
||||||
|
@ -33,33 +32,16 @@ class ServiceEntity(EntityBase):
|
||||||
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(
|
program: Mapped[ProgramTypeEnum] = mapped_column(Enum(ProgramTypeEnum), nullable=False)
|
||||||
Enum(ProgramTypeEnum), nullable=False
|
|
||||||
)
|
|
||||||
|
|
||||||
# relationships
|
# relationships
|
||||||
service_tags: Mapped[list["ServiceTagEntity"]] = relationship(
|
serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(
|
||||||
back_populates="service", cascade="all,delete-orphan"
|
back_populates="service", cascade="all,delete"
|
||||||
)
|
)
|
||||||
|
|
||||||
def to_model(self) -> Service:
|
def to_model(self) -> Service:
|
||||||
return Service(
|
return Service(id=self.id, name=self.name, status=self.status, summary=self.summary, requirements=self.requirements, program=self.program)
|
||||||
id=self.id,
|
|
||||||
name=self.name,
|
|
||||||
status=self.status,
|
|
||||||
summary=self.summary,
|
|
||||||
requirements=self.requirements,
|
|
||||||
program=self.program,
|
|
||||||
tags=[tag.tag.to_model() for tag in self.service_tags],
|
|
||||||
)
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_model(cls, model: Service) -> Self:
|
def from_model(cls, model:Service) -> Self:
|
||||||
return cls(
|
return cls(id=model.id, name=model.name, status=model.status, summary=model.summary, requirements=model.requirements, program=model.program)
|
||||||
id=model.id,
|
|
||||||
name=model.name,
|
|
||||||
status=model.status,
|
|
||||||
summary=model.summary,
|
|
||||||
requirements=model.requirements,
|
|
||||||
program=model.program,
|
|
||||||
)
|
|
|
@ -9,10 +9,6 @@ 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
|
||||||
|
|
||||||
from typing import Self
|
|
||||||
|
|
||||||
from ..models.service_tag_model import ServiceTag
|
|
||||||
|
|
||||||
|
|
||||||
class ServiceTagEntity(EntityBase):
|
class ServiceTagEntity(EntityBase):
|
||||||
|
|
||||||
|
@ -25,16 +21,5 @@ class ServiceTagEntity(EntityBase):
|
||||||
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
|
tagId: Mapped[int] = mapped_column(ForeignKey("tag.id"))
|
||||||
|
|
||||||
# relationships
|
# relationships
|
||||||
service: Mapped["ServiceEntity"] = relationship(back_populates="service_tags")
|
service: Mapped["ServiceEntity"] = relationship(back_populates="serviceTags")
|
||||||
tag: Mapped["TagEntity"] = relationship(back_populates="service")
|
tag: Mapped["TagEntity"] = relationship(back_populates="serviceTags")
|
||||||
|
|
||||||
def to_model(self) -> ServiceTag:
|
|
||||||
return ServiceTag(id=self.id, serviceId=self.serviceId, tagId=self.tagId)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_model(cls, model: "ServiceTag") -> Self:
|
|
||||||
return cls(
|
|
||||||
id=model.id,
|
|
||||||
serviceId=model.serviceId,
|
|
||||||
tagId=model.tagId,
|
|
||||||
)
|
|
||||||
|
|
|
@ -16,25 +16,22 @@ 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
|
||||||
resource: Mapped[list["ResourceTagEntity"]] = relationship(
|
resourceTags: Mapped[list["ResourceTagEntity"]] = relationship(back_populates="tag", cascade="all,delete")
|
||||||
back_populates="tag", cascade="all,delete", overlaps="tag"
|
serviceTags: Mapped[list["ServiceTagEntity"]] = relationship(back_populates="tag", cascade="all,delete")
|
||||||
)
|
|
||||||
service: Mapped[list["ServiceTagEntity"]] = relationship(
|
|
||||||
back_populates="tag", cascade="all,delete", overlaps="tag"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_model(cls, model: Tag) -> Self:
|
def from_model(cls, model: Tag) -> Self:
|
||||||
"""
|
"""
|
||||||
|
@ -45,10 +42,10 @@ class TagEntity(EntityBase):
|
||||||
Returns:
|
Returns:
|
||||||
self: The entity
|
self: The entity
|
||||||
"""
|
"""
|
||||||
|
|
||||||
return cls(
|
return cls(
|
||||||
id=model.id,
|
id=model.id,
|
||||||
content=model.content,
|
content=model.id,
|
||||||
)
|
)
|
||||||
|
|
||||||
def to_model(self) -> Tag:
|
def to_model(self) -> Tag:
|
||||||
|
@ -59,4 +56,10 @@ class TagEntity(EntityBase):
|
||||||
User: A User model for API usage
|
User: A User model for API usage
|
||||||
"""
|
"""
|
||||||
|
|
||||||
return Tag(id=self.id, content=self.content, created_at=self.created_at)
|
return Tag(
|
||||||
|
id=self.id,
|
||||||
|
content=self.content,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -1,7 +0,0 @@
|
||||||
from .user_model import User
|
|
||||||
from .resource_model import Resource
|
|
||||||
from .service_model import Service
|
|
||||||
from .tag_model import Tag
|
|
||||||
from .enum_for_models import ProgramTypeEnum, UserTypeEnum
|
|
||||||
from .resource_tag_model import ResourceTag
|
|
||||||
from .service_tag_model import ServiceTag
|
|
|
@ -4,7 +4,6 @@ from typing import List
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from .enum_for_models import ProgramTypeEnum
|
from .enum_for_models import ProgramTypeEnum
|
||||||
from .tag_model import Tag
|
|
||||||
|
|
||||||
|
|
||||||
class Resource(BaseModel):
|
class Resource(BaseModel):
|
||||||
|
@ -14,4 +13,3 @@ class Resource(BaseModel):
|
||||||
link: str = Field(..., max_length=150, description="link to the resource")
|
link: str = Field(..., max_length=150, description="link to the resource")
|
||||||
program: ProgramTypeEnum
|
program: ProgramTypeEnum
|
||||||
created_at: Optional[datetime]
|
created_at: Optional[datetime]
|
||||||
tags: List[Tag] = []
|
|
||||||
|
|
|
@ -1,7 +1,13 @@
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel, Field
|
||||||
|
from enum import Enum
|
||||||
|
from typing import List
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Optional
|
||||||
|
from .tag_model import Tag
|
||||||
|
from .resource_model import Resource
|
||||||
|
|
||||||
|
|
||||||
class ResourceTag(BaseModel):
|
class ResourceTag(Resource, BaseModel):
|
||||||
id: int | None = None
|
id: int | None = None
|
||||||
tagId: int
|
resourceid: int | None = None
|
||||||
resourceId: int
|
tagid: List[Tag]
|
||||||
|
|
|
@ -4,7 +4,6 @@ from typing import List
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from .enum_for_models import ProgramTypeEnum
|
from .enum_for_models import ProgramTypeEnum
|
||||||
from .tag_model import Tag
|
|
||||||
|
|
||||||
|
|
||||||
class Service(BaseModel):
|
class Service(BaseModel):
|
||||||
|
@ -15,4 +14,3 @@ class Service(BaseModel):
|
||||||
summary: str
|
summary: str
|
||||||
requirements: List[str]
|
requirements: List[str]
|
||||||
program: ProgramTypeEnum
|
program: ProgramTypeEnum
|
||||||
tags: List[Tag] = []
|
|
||||||
|
|
|
@ -4,12 +4,16 @@ from typing import List
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
|
from .enum_for_models import ProgramTypeEnum
|
||||||
|
from .enum_for_models import UserTypeEnum
|
||||||
|
from .service_model import Service
|
||||||
|
|
||||||
from .tag_model import Tag
|
from .tag_model import Tag
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
||||||
|
|
||||||
class ServiceTag(BaseModel):
|
class ServiceTag(Service, BaseModel):
|
||||||
id: int | None = None
|
id: int | None = None
|
||||||
serviceId: int | None = None
|
serviceid: int | None = None
|
||||||
tagId: int
|
tagId: List[Tag]
|
||||||
|
|
|
@ -6,12 +6,7 @@ from ..database import engine, _engine_str
|
||||||
from ..env import getenv
|
from ..env import getenv
|
||||||
from .. import entities
|
from .. import entities
|
||||||
|
|
||||||
from ..test.services import (
|
from ..test.services import user_test_data, service_test_data, resource_test_data
|
||||||
user_test_data,
|
|
||||||
service_test_data,
|
|
||||||
resource_test_data,
|
|
||||||
tag_test_data,
|
|
||||||
)
|
|
||||||
|
|
||||||
database = getenv("POSTGRES_DATABASE")
|
database = getenv("POSTGRES_DATABASE")
|
||||||
|
|
||||||
|
@ -28,6 +23,5 @@ entities.EntityBase.metadata.create_all(engine)
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
user_test_data.insert_test_data(session)
|
user_test_data.insert_test_data(session)
|
||||||
service_test_data.insert_test_data(session)
|
service_test_data.insert_test_data(session)
|
||||||
tag_test_data.insert_test_data(session)
|
|
||||||
resource_test_data.insert_test_data(session)
|
resource_test_data.insert_test_data(session)
|
||||||
session.commit()
|
session.commit()
|
||||||
|
|
|
@ -1,6 +1,4 @@
|
||||||
from fastapi import Depends
|
from fastapi import Depends
|
||||||
|
|
||||||
from backend.services.tag import TagService
|
|
||||||
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 select
|
||||||
|
@ -8,18 +6,13 @@ 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 ProgramNotAssignedException, ResourceNotFoundException
|
from .exceptions import ResourceNotFoundException
|
||||||
|
|
||||||
|
|
||||||
class ResourceService:
|
class ResourceService:
|
||||||
|
|
||||||
def __init__(
|
def __init__(self, session: Session = Depends(db_session)):
|
||||||
self,
|
|
||||||
session: Session = Depends(db_session),
|
|
||||||
tag_service: TagService = Depends(),
|
|
||||||
):
|
|
||||||
self._session = session
|
self._session = session
|
||||||
self._tag_service = tag_service
|
|
||||||
|
|
||||||
def get_resource_by_user(self, subject: User):
|
def get_resource_by_user(self, subject: User):
|
||||||
"""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"""
|
||||||
|
@ -50,22 +43,15 @@ class ResourceService:
|
||||||
Returns:
|
Returns:
|
||||||
Resource: Object added to table
|
Resource: Object added to table
|
||||||
"""
|
"""
|
||||||
if user.role != UserTypeEnum.ADMIN:
|
if resource.role != user.role or resource.group != user.group:
|
||||||
raise ProgramNotAssignedException(
|
raise PermissionError(
|
||||||
f"User is not {UserTypeEnum.ADMIN}, cannot update resource"
|
"User does not have permission to add resources in this role or group."
|
||||||
)
|
)
|
||||||
|
|
||||||
resource_entity = ResourceEntity.from_model(resource)
|
resource_entity = ResourceEntity.from_model(resource)
|
||||||
self._session.add(resource_entity)
|
self._session.add(resource_entity)
|
||||||
self._session.flush()
|
|
||||||
|
|
||||||
for tag in resource.tags:
|
|
||||||
tag_entity = self._tag_service.get_or_create_tag(tag.content)
|
|
||||||
self._tag_service.add_tag_resource(
|
|
||||||
user, tag_entity, ResourceEntity.to_model(resource_entity)
|
|
||||||
)
|
|
||||||
|
|
||||||
self._session.commit()
|
self._session.commit()
|
||||||
|
|
||||||
return resource_entity.to_model()
|
return resource_entity.to_model()
|
||||||
|
|
||||||
def get_by_id(self, user: User, id: int) -> Resource:
|
def get_by_id(self, user: User, id: int) -> Resource:
|
||||||
|
@ -86,7 +72,8 @@ class ResourceService:
|
||||||
self._session.query(ResourceEntity)
|
self._session.query(ResourceEntity)
|
||||||
.filter(
|
.filter(
|
||||||
ResourceEntity.id == id,
|
ResourceEntity.id == id,
|
||||||
ResourceEntity.program.in_(user.program),
|
ResourceEntity.role == user.role,
|
||||||
|
ResourceEntity.group == user.group,
|
||||||
)
|
)
|
||||||
.one_or_none()
|
.one_or_none()
|
||||||
)
|
)
|
||||||
|
@ -110,38 +97,22 @@ class ResourceService:
|
||||||
Raises:
|
Raises:
|
||||||
ResourceNotFoundException: If no resource is found with the corresponding ID
|
ResourceNotFoundException: If no resource is found with the corresponding ID
|
||||||
"""
|
"""
|
||||||
if user.role != UserTypeEnum.ADMIN:
|
if resource.role != user.role or resource.group != user.group:
|
||||||
raise ProgramNotAssignedException(
|
raise PermissionError(
|
||||||
f"User is not {UserTypeEnum.ADMIN}, cannot update service"
|
"User does not have permission to update this resource."
|
||||||
)
|
)
|
||||||
|
|
||||||
resource_entity = (
|
obj = self._session.get(ResourceEntity, resource.id) if resource.id else None
|
||||||
self._session.query(ResourceEntity)
|
|
||||||
.filter(ResourceEntity.id == resource.id)
|
|
||||||
.one_or_none()
|
|
||||||
)
|
|
||||||
|
|
||||||
self._tag_service.delete_all_tags_service(resource)
|
if obj is None:
|
||||||
|
|
||||||
if resource_entity is None:
|
|
||||||
raise ResourceNotFoundException(
|
raise ResourceNotFoundException(
|
||||||
"The resource you are searching for does not exist."
|
f"No resource found with matching id: {resource.id}"
|
||||||
)
|
|
||||||
|
|
||||||
resource_entity.name = resource.name
|
|
||||||
resource_entity.status = resource.status
|
|
||||||
resource_entity.summary = resource.summary
|
|
||||||
resource_entity.requirements = resource.requirements
|
|
||||||
resource_entity.program = resource.program
|
|
||||||
|
|
||||||
for tag in resource.tags:
|
|
||||||
tag_entity = self._tag_service.get_or_create_tag(tag.content)
|
|
||||||
self._tag_service.add_tag_service(
|
|
||||||
user, tag_entity, ResourceEntity.to_model(resource_entity)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
obj.update_from_model(resource) # Assuming an update method exists
|
||||||
self._session.commit()
|
self._session.commit()
|
||||||
return resource_entity.to_model()
|
|
||||||
|
return obj.to_model()
|
||||||
|
|
||||||
def delete(self, user: User, id: int) -> None:
|
def delete(self, user: User, id: int) -> None:
|
||||||
"""
|
"""
|
||||||
|
|
|
@ -1,9 +1,5 @@
|
||||||
from typing import List
|
|
||||||
from fastapi import Depends
|
from fastapi import Depends
|
||||||
|
|
||||||
from backend.entities.service_tag_entity import ServiceTagEntity
|
|
||||||
from backend.entities.tag_entity import TagEntity
|
|
||||||
|
|
||||||
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_
|
||||||
|
@ -16,19 +12,12 @@ from backend.services.exceptions import (
|
||||||
ServiceNotFoundException,
|
ServiceNotFoundException,
|
||||||
ProgramNotAssignedException,
|
ProgramNotAssignedException,
|
||||||
)
|
)
|
||||||
from . import TagService
|
|
||||||
from ..models import Tag
|
|
||||||
|
|
||||||
|
|
||||||
class ServiceService:
|
class ServiceService:
|
||||||
|
|
||||||
def __init__(
|
def __init__(self, session: Session = Depends(db_session)):
|
||||||
self,
|
|
||||||
session: Session = Depends(db_session),
|
|
||||||
tag_service: TagService = Depends(),
|
|
||||||
):
|
|
||||||
self._session = session
|
self._session = session
|
||||||
self._tag_service = tag_service
|
|
||||||
|
|
||||||
def get_service_by_program(self, program: ProgramTypeEnum) -> list[Service]:
|
def get_service_by_program(self, program: ProgramTypeEnum) -> list[Service]:
|
||||||
"""Service method getting services belonging to a particular program."""
|
"""Service method getting services belonging to a particular program."""
|
||||||
|
@ -44,8 +33,8 @@ class ServiceService:
|
||||||
|
|
||||||
if entity is None:
|
if entity is None:
|
||||||
raise ServiceNotFoundException(f"Service with id: {id} does not exist")
|
raise ServiceNotFoundException(f"Service with id: {id} does not exist")
|
||||||
service = entity.to_model()
|
|
||||||
return service
|
return entity.to_model()
|
||||||
|
|
||||||
def get_service_by_name(self, name: str) -> Service:
|
def get_service_by_name(self, name: str) -> Service:
|
||||||
"""Service method getting services by id."""
|
"""Service method getting services by id."""
|
||||||
|
@ -55,9 +44,7 @@ class ServiceService:
|
||||||
if entity is None:
|
if entity is None:
|
||||||
raise ServiceNotFoundException(f"Service with name: {name} does not exist")
|
raise ServiceNotFoundException(f"Service with name: {name} does not exist")
|
||||||
|
|
||||||
service = entity.to_model()
|
return entity.to_model()
|
||||||
# service.tags.extend(TagService.get_tags_for_service(TagService, service))
|
|
||||||
return service
|
|
||||||
|
|
||||||
def get_service_by_user(self, subject: User):
|
def get_service_by_user(self, subject: User):
|
||||||
"""Service method getting all of the services that a user has access to based on role"""
|
"""Service method getting all of the services that a user has access to based on role"""
|
||||||
|
@ -86,8 +73,8 @@ class ServiceService:
|
||||||
|
|
||||||
query = select(ServiceEntity)
|
query = select(ServiceEntity)
|
||||||
entities = self._session.scalars(query).all()
|
entities = self._session.scalars(query).all()
|
||||||
services = [service.to_model() for service in entities]
|
|
||||||
return services
|
return [service.to_model() for service in entities]
|
||||||
|
|
||||||
def create(self, subject: User, service: Service) -> Service:
|
def create(self, subject: User, service: Service) -> Service:
|
||||||
"""Creates/adds a service to the table."""
|
"""Creates/adds a service to the table."""
|
||||||
|
@ -98,14 +85,6 @@ class ServiceService:
|
||||||
|
|
||||||
service_entity = ServiceEntity.from_model(service)
|
service_entity = ServiceEntity.from_model(service)
|
||||||
self._session.add(service_entity)
|
self._session.add(service_entity)
|
||||||
self._session.flush()
|
|
||||||
|
|
||||||
for tag in service.tags:
|
|
||||||
tag_entity = self._tag_service.get_or_create_tag(tag.content)
|
|
||||||
self._tag_service.add_tag_service(
|
|
||||||
subject, tag_entity, ServiceEntity.to_model(service_entity)
|
|
||||||
)
|
|
||||||
|
|
||||||
self._session.commit()
|
self._session.commit()
|
||||||
return service_entity.to_model()
|
return service_entity.to_model()
|
||||||
|
|
||||||
|
@ -116,13 +95,7 @@ class ServiceService:
|
||||||
f"User is not {UserTypeEnum.ADMIN}, cannot update service"
|
f"User is not {UserTypeEnum.ADMIN}, cannot update service"
|
||||||
)
|
)
|
||||||
|
|
||||||
service_entity = (
|
service_entity = self._session.get(ServiceEntity, service.id)
|
||||||
self._session.query(ServiceEntity)
|
|
||||||
.filter(ServiceEntity.id == service.id)
|
|
||||||
.one_or_none()
|
|
||||||
)
|
|
||||||
|
|
||||||
self._tag_service.delete_all_tags_service(service)
|
|
||||||
|
|
||||||
if service_entity is None:
|
if service_entity is None:
|
||||||
raise ServiceNotFoundException(
|
raise ServiceNotFoundException(
|
||||||
|
@ -135,49 +108,20 @@ class ServiceService:
|
||||||
service_entity.requirements = service.requirements
|
service_entity.requirements = service.requirements
|
||||||
service_entity.program = service.program
|
service_entity.program = service.program
|
||||||
|
|
||||||
for tag in service.tags:
|
|
||||||
tag_entity = self._tag_service.get_or_create_tag(tag.content)
|
|
||||||
self._tag_service.add_tag_service(
|
|
||||||
subject, tag_entity, ServiceEntity.to_model(service_entity)
|
|
||||||
)
|
|
||||||
|
|
||||||
self._session.commit()
|
self._session.commit()
|
||||||
|
|
||||||
return service_entity.to_model()
|
return service_entity.to_model()
|
||||||
|
|
||||||
def delete(self, subject: User, service: Service) -> None:
|
def delete(self, subject: User, service: Service) -> None:
|
||||||
"""Deletes a service from the table."""
|
"""Deletes a service from the table."""
|
||||||
if subject.role != UserTypeEnum.ADMIN:
|
if subject.role != UserTypeEnum.ADMIN:
|
||||||
raise ProgramNotAssignedException(f"User is not {UserTypeEnum.ADMIN}")
|
raise ProgramNotAssignedException(f"User is not {UserTypeEnum.ADMIN}")
|
||||||
service_entity = (
|
service_entity = self._session.get(ServiceEntity, service.id)
|
||||||
self._session.query(ServiceEntity)
|
|
||||||
.filter(ServiceEntity.id == service.id)
|
|
||||||
.one_or_none()
|
|
||||||
)
|
|
||||||
if service_entity is None:
|
if service_entity is None:
|
||||||
raise ServiceNotFoundException(
|
raise ServiceNotFoundException(
|
||||||
"The service you are searching for does not exist."
|
"The service you are searching for does not exist."
|
||||||
)
|
)
|
||||||
self._tag_service.delete_all_tags_service(service_entity.to_model())
|
|
||||||
self._session.delete(service_entity)
|
self._session.delete(service_entity)
|
||||||
self._session.commit()
|
self._session.commit()
|
||||||
|
|
||||||
def add_tag(self, subject: User, service: Service, tag: Tag):
|
|
||||||
service = self.get_service_by_id(service.id)
|
|
||||||
tag = self._tag_service.get_tag_by_id(tag.id)
|
|
||||||
self._tag_service.add_tag_service(subject, service.id, tag.id)
|
|
||||||
|
|
||||||
def remove_tag(self, subject: User, service: Service, tag: Tag) -> None:
|
|
||||||
service_tag = (
|
|
||||||
self._session.query(ServiceTagEntity)
|
|
||||||
.filter(
|
|
||||||
ServiceTagEntity.serviceId == service.id,
|
|
||||||
ServiceTagEntity.tagId == tag.id,
|
|
||||||
)
|
|
||||||
.one_or_none()
|
|
||||||
)
|
|
||||||
if service_tag is None:
|
|
||||||
raise Exception(
|
|
||||||
f"No tag with id {tag.id} found for service with id {service.id}."
|
|
||||||
)
|
|
||||||
self._session.delete(service_tag)
|
|
||||||
self._session.commit()
|
|
||||||
|
|
|
@ -1,13 +1,15 @@
|
||||||
from fastapi import Depends
|
from fastapi import Depends
|
||||||
|
|
||||||
from backend.models.enum_for_models import UserTypeEnum
|
|
||||||
from ..database import db_session
|
from ..database import db_session
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
|
from ..models.tag_model import Tag
|
||||||
|
from ..entities.tag_entity import TagEntity
|
||||||
|
from ..entities.resource_tag_entity import ResourceTagEntity
|
||||||
|
from ..entities.service_tag_entity import ServiceTagEntity
|
||||||
|
from .exceptions import ResourceNotFoundException
|
||||||
|
from ..models.user_model import User
|
||||||
|
from ..models.resource_model import Resource
|
||||||
|
from ..models.service_model import Service
|
||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
from ..entities import TagEntity, ResourceTagEntity, ServiceTagEntity
|
|
||||||
from ..models import User, Resource, Service, Tag, ResourceTag, ServiceTag
|
|
||||||
from .exceptions import ProgramNotAssignedException, ResourceNotFoundException
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
|
|
||||||
class TagService:
|
class TagService:
|
||||||
|
@ -18,33 +20,39 @@ class TagService:
|
||||||
def all(self) -> list[Tag]:
|
def all(self) -> list[Tag]:
|
||||||
"""Returns a list of all Tags"""
|
"""Returns a list of all Tags"""
|
||||||
|
|
||||||
query = select(TagEntity)
|
query = select(TagEntity)
|
||||||
entities = self._session.scalars(query).all()
|
entities = self._session.scalars(query).all()
|
||||||
|
|
||||||
return [entity.to_model() for entity in entities]
|
return [entity.to_model() for entity in entities]
|
||||||
|
|
||||||
def get_tag_by_id(self, id: int) -> Tag:
|
def get_tag_by_id(self, id:int) -> Tag:
|
||||||
"""Returns tag based on it's id."""
|
"""Returns tag based on it's id."""
|
||||||
|
|
||||||
tag = self._session.query(TagEntity).filter(TagEntity.id == id).one_or_none()
|
tag = self._session.query(TagEntity).filter(TagEntity.id == id).one_or_none()
|
||||||
return tag
|
return tag.to_model()
|
||||||
|
|
||||||
def create(self, tag: Tag) -> Tag:
|
def create(self, user: User, tag: Tag) -> Tag:
|
||||||
"""Creates a tag in the database with id and content."""
|
"""Creates a tag in the database with id and content."""
|
||||||
|
|
||||||
# Add user permission check here if needed
|
# Add user permission check here if needed
|
||||||
|
|
||||||
tag_entity = TagEntity.from_model(tag)
|
tag_entity = TagEntity.from_model(tag)
|
||||||
self._session.add(tag_entity)
|
self._session.add(tag_entity)
|
||||||
self._session.commit()
|
self._session.commit
|
||||||
|
|
||||||
return tag_entity.to_model()
|
return tag_entity.to_model()
|
||||||
|
|
||||||
def delete(self, id: int) -> None:
|
def delete(self, user: User, id:int) -> None:
|
||||||
"""Method to delete a tag from the database, along with all connections."""
|
"""Method to delete a tag from the database, along with all connections."""
|
||||||
|
|
||||||
tag = self._session.query(TagEntity).filter(TagEntity.id == id).one_or_none()
|
tag = (
|
||||||
|
self._session.query(TagEntity)
|
||||||
|
.filter(
|
||||||
|
TagEntity.id == id
|
||||||
|
)
|
||||||
|
.one_or_none()
|
||||||
|
)
|
||||||
|
|
||||||
if tag is None:
|
if tag is None:
|
||||||
raise ResourceNotFoundException(f"No tag found with matching id: {id}")
|
raise ResourceNotFoundException(f"No tag found with matching id: {id}")
|
||||||
|
|
||||||
|
@ -52,16 +60,20 @@ class TagService:
|
||||||
|
|
||||||
resource_tags = (
|
resource_tags = (
|
||||||
self._session.query(ResourceTagEntity)
|
self._session.query(ResourceTagEntity)
|
||||||
.filter(ResourceTagEntity.tagId == id)
|
.filter(
|
||||||
|
ResourceTagEntity.tagId == id
|
||||||
|
)
|
||||||
.all()
|
.all()
|
||||||
)
|
)
|
||||||
|
|
||||||
for tag in resource_tags:
|
for tag in resource_tags:
|
||||||
self._session.delete(tag)
|
self._session.delete(tag)
|
||||||
|
|
||||||
service_tags = (
|
service_tags = (
|
||||||
self._session.query(ServiceTagEntity)
|
self._session.query(ServiceTagEntity)
|
||||||
.filter(ServiceTagEntity.tagId == id)
|
.filter(
|
||||||
|
ServiceTagEntity.tagId == id
|
||||||
|
)
|
||||||
.all()
|
.all()
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -69,147 +81,42 @@ class TagService:
|
||||||
self._session.delete(tag)
|
self._session.delete(tag)
|
||||||
|
|
||||||
self._session.commit()
|
self._session.commit()
|
||||||
|
|
||||||
def get_tags_for_resource(self, resource: Resource) -> list[Tag]:
|
def get_tags_for_resource(self, user: User, resource: Resource) -> list[Tag]:
|
||||||
"""Get tags based on a resource."""
|
"""Get tags based on a resource."""
|
||||||
tags: list[Tag] = []
|
tags: list[Tag]
|
||||||
resource_tags = (
|
resource_tags = (
|
||||||
self._session.query(ResourceTagEntity)
|
self._session.query(ResourceTagEntity)
|
||||||
.filter(ResourceTagEntity.tagId == resource.id)
|
.filter(
|
||||||
|
ResourceTagEntity.tagId == resource.id
|
||||||
|
)
|
||||||
.all()
|
.all()
|
||||||
)
|
)
|
||||||
|
|
||||||
if resource_tags is None:
|
if resource_tags is None:
|
||||||
raise ResourceNotFoundException(
|
raise ResourceNotFoundException(f"No tags found for resource with id: {resource.id}")
|
||||||
f"No tags found for resource with id: {resource.id}"
|
|
||||||
)
|
|
||||||
|
|
||||||
for tag in resource_tags:
|
for tag in resource_tags:
|
||||||
tags.append(self.get_tag_by_id(tag.id))
|
tags.append(self.get_tag_by_id(tag.id))
|
||||||
|
|
||||||
return tags
|
return tags
|
||||||
|
|
||||||
def get_tags_for_service(self, service: Service) -> list[Tag]:
|
def get_tags_for_service(self, user: User, service: Service) -> list[Tag]:
|
||||||
"""Get tags based on a resource."""
|
"""Get tags based on a resource."""
|
||||||
tags: list[Tag] = []
|
tags: list[Tag]
|
||||||
service_tags = (
|
service_tags = (
|
||||||
self._session.query(ServiceTagEntity)
|
self._session.query(ServiceTagEntity)
|
||||||
.filter(ServiceTagEntity.serviceId == service.id)
|
.filter(
|
||||||
|
ServiceTagEntity.tagId == service.id
|
||||||
|
)
|
||||||
.all()
|
.all()
|
||||||
)
|
)
|
||||||
|
|
||||||
if service_tags is None:
|
if service_tags is None:
|
||||||
raise ResourceNotFoundException(
|
raise ResourceNotFoundException(f"No tags found for service with id: {service.id}")
|
||||||
f"No tags found for service with id: {service.id}"
|
|
||||||
)
|
|
||||||
|
|
||||||
for tag in service_tags:
|
for tag in service_tags:
|
||||||
tags.append(self.get_tag_by_id(tag.id))
|
tags.append(self.get_tag_by_id(tag.id))
|
||||||
|
|
||||||
return tags
|
return tags
|
||||||
|
|
||||||
def add_tag_resource(self, user: User, tag: Tag, resource: Resource) -> None:
|
|
||||||
"""Adds a tag to a resource"""
|
|
||||||
if user.role != UserTypeEnum.ADMIN:
|
|
||||||
raise ProgramNotAssignedException(
|
|
||||||
f"User is not {UserTypeEnum.ADMIN}, cannot update resource tags"
|
|
||||||
)
|
|
||||||
|
|
||||||
existing_tag = (
|
|
||||||
self._session.query(ResourceTagEntity)
|
|
||||||
.filter(
|
|
||||||
ResourceTagEntity.tagId == tag.id,
|
|
||||||
ResourceTagEntity.resourceId == resource.id,
|
|
||||||
)
|
|
||||||
.first()
|
|
||||||
)
|
|
||||||
|
|
||||||
if existing_tag:
|
|
||||||
raise Exception(
|
|
||||||
f"Tag with id {tag.id} already exists for resource with id {resource.id}."
|
|
||||||
)
|
|
||||||
|
|
||||||
resource_tag_entity = ResourceTagEntity.from_model(
|
|
||||||
ResourceTag(tagId=tag.id, resourceId=resource.id)
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
|
||||||
self._session.add(resource_tag_entity)
|
|
||||||
self._session.commit()
|
|
||||||
except Exception as e:
|
|
||||||
self._session.rollback() # Rollback in case of error
|
|
||||||
raise Exception("Failed to add tag to resource") from e
|
|
||||||
|
|
||||||
def add_tag_service(self, user: User, tag: Tag, service: Service) -> None:
|
|
||||||
"""Adds a tag to a service"""
|
|
||||||
|
|
||||||
existing_tag = (
|
|
||||||
self._session.query(ServiceTagEntity)
|
|
||||||
.filter(
|
|
||||||
ServiceTagEntity.tagId == tag.id,
|
|
||||||
ServiceTagEntity.serviceId == service.id,
|
|
||||||
)
|
|
||||||
.first()
|
|
||||||
)
|
|
||||||
|
|
||||||
if existing_tag:
|
|
||||||
raise Exception(
|
|
||||||
f"Tag with id {tag.id} already exists for service with id {service.id}."
|
|
||||||
)
|
|
||||||
|
|
||||||
service_tag_entity = ServiceTagEntity.from_model(
|
|
||||||
ServiceTag(tagId=tag.id, serviceId=service.id)
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
|
||||||
self._session.add(service_tag_entity)
|
|
||||||
self._session.commit()
|
|
||||||
except Exception as e:
|
|
||||||
self._session.rollback() # Rollback in case of error
|
|
||||||
raise Exception("Failed to add tag to service") from e
|
|
||||||
|
|
||||||
def delete_all_tags_service(self, service: Service) -> None:
|
|
||||||
"""Deletes all service tags for a service"""
|
|
||||||
|
|
||||||
service_tags = self._session.query(ServiceTagEntity).filter(
|
|
||||||
ServiceTagEntity.serviceId == service.id
|
|
||||||
)
|
|
||||||
|
|
||||||
if service_tags.count() == 0:
|
|
||||||
raise ResourceNotFoundException
|
|
||||||
|
|
||||||
service_tags.delete(synchronize_session=False)
|
|
||||||
self._session.commit()
|
|
||||||
|
|
||||||
def delete_all_tags_resource(self, resource: Resource) -> None:
|
|
||||||
"""Deletes all resource tags for a resource"""
|
|
||||||
|
|
||||||
resource_tags = self._session.query(ResourceTagEntity).filter(
|
|
||||||
ResourceTagEntity.resourceId == resource.id
|
|
||||||
)
|
|
||||||
|
|
||||||
if resource_tags.count() == 0:
|
|
||||||
raise ResourceNotFoundException
|
|
||||||
|
|
||||||
resource_tags.delete(synchronize_session=False)
|
|
||||||
self._session.commit()
|
|
||||||
|
|
||||||
def get_or_create_tag(self, content: str) -> Tag:
|
|
||||||
existing_tag = (
|
|
||||||
self._session.query(TagEntity)
|
|
||||||
.filter(TagEntity.content == content)
|
|
||||||
.one_or_none()
|
|
||||||
)
|
|
||||||
|
|
||||||
if existing_tag:
|
|
||||||
return existing_tag.to_model()
|
|
||||||
|
|
||||||
try:
|
|
||||||
tag = Tag(content=content)
|
|
||||||
tag_entity = TagEntity.from_model(tag)
|
|
||||||
self._session.add(tag_entity)
|
|
||||||
self._session.commit()
|
|
||||||
return tag_entity.to_model()
|
|
||||||
except Exception as e:
|
|
||||||
self._session.rollback()
|
|
||||||
raise Exception(f"Failed to create tag with content: {tag.content}") from e
|
|
||||||
|
|
|
@ -4,12 +4,7 @@ import pytest
|
||||||
from sqlalchemy import Engine, create_engine, text
|
from sqlalchemy import Engine, create_engine, text
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
from sqlalchemy.exc import OperationalError
|
from sqlalchemy.exc import OperationalError
|
||||||
from .services import (
|
from .services import user_test_data, tag_test_data, service_test_data
|
||||||
user_test_data,
|
|
||||||
tag_test_data,
|
|
||||||
service_test_data,
|
|
||||||
resource_test_data,
|
|
||||||
)
|
|
||||||
|
|
||||||
from ..database import _engine_str
|
from ..database import _engine_str
|
||||||
from ..env import getenv
|
from ..env import getenv
|
||||||
|
@ -62,6 +57,5 @@ def setup_insert_data_fixture(session: Session):
|
||||||
user_test_data.insert_fake_data(session)
|
user_test_data.insert_fake_data(session)
|
||||||
tag_test_data.insert_fake_data(session)
|
tag_test_data.insert_fake_data(session)
|
||||||
service_test_data.insert_fake_data(session)
|
service_test_data.insert_fake_data(session)
|
||||||
resource_test_data.insert_fake_data(session)
|
|
||||||
session.commit()
|
session.commit()
|
||||||
yield
|
yield
|
||||||
|
|
|
@ -3,7 +3,11 @@
|
||||||
import pytest
|
import pytest
|
||||||
from unittest.mock import create_autospec
|
from unittest.mock import create_autospec
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
from ...services import UserService, TagService, ServiceService, ResourceService
|
from ...services import UserService
|
||||||
|
from ...services import TagService
|
||||||
|
from ...services import ServiceService
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture()
|
@pytest.fixture()
|
||||||
|
@ -11,20 +15,12 @@ def user_svc(session: Session):
|
||||||
"""This fixture is used to test the UserService class"""
|
"""This fixture is used to test the UserService class"""
|
||||||
return UserService(session)
|
return UserService(session)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture()
|
@pytest.fixture()
|
||||||
def tag_svc(session: Session):
|
def tag_svc(session: Session):
|
||||||
"""This fixture is used to test the TagService class"""
|
"""This fixture is used to test the TagService class"""
|
||||||
return TagService(session)
|
return TagService(session)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture()
|
@pytest.fixture()
|
||||||
def service_svc(session: Session):
|
def service_svc(session: Session):
|
||||||
"""This fixture is used to test the ServiceService class"""
|
"""This fixture is used to test the ServiceService class"""
|
||||||
return ServiceService(session)
|
return ServiceService(session)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture()
|
|
||||||
def resource_svc(session: Session):
|
|
||||||
"""This fixutre is used to test the ResourceService class"""
|
|
||||||
return ResourceService(session)
|
|
|
@ -1,10 +0,0 @@
|
||||||
from backend.services import ResourceService, TagService
|
|
||||||
from .user_test_data import admin
|
|
||||||
from .fixtures import resource_svc, tag_svc
|
|
||||||
|
|
||||||
|
|
||||||
def test_temp(resource_svc: ResourceService, tag_svc: TagService):
|
|
||||||
resources = resource_svc.get_resource_by_user(admin)
|
|
||||||
tags = tag_svc.all()
|
|
||||||
print(tags)
|
|
||||||
print(resources)
|
|
|
@ -3,8 +3,7 @@ from sqlalchemy.orm import Session
|
||||||
|
|
||||||
from ...entities import ServiceEntity
|
from ...entities import ServiceEntity
|
||||||
from ...models.enum_for_models import ProgramTypeEnum
|
from ...models.enum_for_models import ProgramTypeEnum
|
||||||
from ...models import Service, Tag
|
from ...models.service_model import Service
|
||||||
from .tag_test_data import tags
|
|
||||||
|
|
||||||
service1 = Service(
|
service1 = Service(
|
||||||
id=1,
|
id=1,
|
||||||
|
@ -96,7 +95,6 @@ service_1 = Service(
|
||||||
summary="24/7 support for individuals in crisis",
|
summary="24/7 support for individuals in crisis",
|
||||||
requirements=["Anonymous", "Confidential"],
|
requirements=["Anonymous", "Confidential"],
|
||||||
program=ProgramTypeEnum.DOMESTIC,
|
program=ProgramTypeEnum.DOMESTIC,
|
||||||
tags=[tags[0], tags[1]],
|
|
||||||
)
|
)
|
||||||
|
|
||||||
service_2 = Service(
|
service_2 = Service(
|
||||||
|
|
|
@ -2,55 +2,13 @@
|
||||||
|
|
||||||
# PyTest
|
# PyTest
|
||||||
import pytest
|
import pytest
|
||||||
from ...services import TagService, ResourceService, ServiceService
|
from ...services.tag import TagService
|
||||||
from .fixtures import tag_svc, resource_svc, service_svc
|
from .fixtures import tag_svc
|
||||||
from .tag_test_data import tag_to_create, tag_to_create_no_id, tags
|
from .tag_test_data import tag1, tag2, tag3
|
||||||
from .user_test_data import admin
|
from . import tag_test_data
|
||||||
|
|
||||||
|
|
||||||
def test_get_all(tag_svc: TagService):
|
def test_get_all(tag_svc: TagService):
|
||||||
"""Test that all tags can be retrieved."""
|
"""Test that all tags can be retrieved."""
|
||||||
all_tags = tag_svc.all()
|
tags = tag_svc.all()
|
||||||
assert len(all_tags) == len(tags)
|
assert len(tags) == 3
|
||||||
|
|
||||||
|
|
||||||
def test_create_tag(tag_svc: TagService):
|
|
||||||
"""Test creation of tag"""
|
|
||||||
created_tag = tag_svc.create(admin, tag_to_create)
|
|
||||||
|
|
||||||
assert created_tag.content == tag_to_create.content
|
|
||||||
assert len(tag_svc.all()) == len(tags) + 1
|
|
||||||
|
|
||||||
|
|
||||||
def test_create_tag_no_id(tag_svc: TagService):
|
|
||||||
"""Test creation of tag"""
|
|
||||||
created_tag = tag_svc.create(admin, tag_to_create_no_id)
|
|
||||||
queried_tag = tag_svc.get_tag_by_id(4)
|
|
||||||
|
|
||||||
assert created_tag.content == tag_to_create_no_id.content
|
|
||||||
assert len(tag_svc.all()) == len(tags) + 1
|
|
||||||
assert queried_tag.content == created_tag.content
|
|
||||||
|
|
||||||
|
|
||||||
def test_resource_tag_creation(tag_svc: TagService, resource_svc: ResourceService):
|
|
||||||
"""Test creation and attachment of resource tag"""
|
|
||||||
|
|
||||||
resource = resource_svc.get_resource_by_user(admin)[0]
|
|
||||||
tag_svc.add_tag_resource(admin, tags[0], resource)
|
|
||||||
updated_resource = resource_svc.get_by_id(admin, resource.id)
|
|
||||||
|
|
||||||
assert len(resource.tags) == 0
|
|
||||||
assert len(updated_resource.tags) == 1
|
|
||||||
assert resource.id == updated_resource.id
|
|
||||||
|
|
||||||
|
|
||||||
def test_service_tag_creation(tag_svc: TagService, service_svc: ServiceService):
|
|
||||||
"""Test creation and attachment of service tag"""
|
|
||||||
|
|
||||||
service = service_svc.get_service_by_user(admin)[0]
|
|
||||||
tag_svc.add_tag_service(admin, tags[0], service)
|
|
||||||
updated_service = service_svc.get_service_by_id(service.id)
|
|
||||||
|
|
||||||
assert len(service.tags) == 0
|
|
||||||
assert len(updated_service.tags) == 1
|
|
||||||
assert service.id == updated_service.id
|
|
|
@ -11,9 +11,7 @@ tag2 = Tag(id=2, content="Tag 2", created_at=datetime.now())
|
||||||
|
|
||||||
tag3 = Tag(id=3, content="Tag 3", created_at=datetime.now())
|
tag3 = Tag(id=3, content="Tag 3", created_at=datetime.now())
|
||||||
|
|
||||||
tag_to_create = Tag(id=4, content="Tag 4", created_at=datetime.now())
|
tagToCreate = Tag(id=4, content="Tag 4", created_at=datetime.now())
|
||||||
|
|
||||||
tag_to_create_no_id = Tag(content="Tag 5", created_at=datetime.now())
|
|
||||||
|
|
||||||
tags = [tag1, tag2, tag3]
|
tags = [tag1, tag2, tag3]
|
||||||
|
|
||||||
|
@ -63,25 +61,6 @@ def insert_fake_data(session: Session):
|
||||||
session.commit()
|
session.commit()
|
||||||
|
|
||||||
|
|
||||||
def insert_test_data(session: Session):
|
|
||||||
"""Inserts test organization data into the test session."""
|
|
||||||
|
|
||||||
global tags
|
|
||||||
|
|
||||||
# Create entities for test organization data
|
|
||||||
entities = []
|
|
||||||
for tag in tags:
|
|
||||||
entity = TagEntity.from_model(tag)
|
|
||||||
session.add(entity)
|
|
||||||
entities.append(entity)
|
|
||||||
|
|
||||||
# Reset table IDs to prevent ID conflicts
|
|
||||||
reset_table_id_seq(session, TagEntity, TagEntity.id, len(tags) + 1)
|
|
||||||
|
|
||||||
# Commit all changes
|
|
||||||
session.commit()
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture(autouse=True)
|
||||||
def fake_data_fixture(session: Session):
|
def fake_data_fixture(session: Session):
|
||||||
"""Insert fake data the session automatically when test is run.
|
"""Insert fake data the session automatically when test is run.
|
||||||
|
|
Loading…
Reference in New Issue
Block a user