Push branch, write user services and start tag service

This commit is contained in:
emmalynf 2024-04-04 22:53:02 -04:00
parent 7a1ec296b7
commit c1f7ff1df3
5 changed files with 248 additions and 25 deletions

View File

@ -20,6 +20,11 @@ from datetime import datetime
from backend.entities.program_enum import ProgramEnum
from .user_enum import RoleEnum
#Import models for User methods
from ..models.user_model import User
from typing import Self
class UserEntity(EntityBase):
"""Serves as the database model for User table"""
@ -30,61 +35,55 @@ class UserEntity(EntityBase):
# set fields or 'columns' for the user table
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
username: Mapped[str] = mapped_column(
String(32), nullable=False, default="", unique=True
)
role: Mapped[RoleEnum] = mapped_column(RoleEnum, nullable=False)
username: Mapped[str] = mapped_column(
String(32), nullable=False, default="", unique=True
)
username: Mapped[str] = mapped_column(String(32), nullable=False, default="", unique=True )
role: Mapped[RoleEnum] = mapped_column(RoleEnum, nullable=False)
email: Mapped[str] = mapped_column(String(50), nullable=False, unique=True)
program: Mapped[list[ProgramEnum]] = mapped_column(
ARRAY(ProgramEnum), nullable=False
)
program: Mapped[list[ProgramEnum]] = mapped_column(
ARRAY(ProgramEnum), nullable=False
)
program: Mapped[list[ProgramEnum]] = mapped_column(ARRAY(ProgramEnum), nullable=False)
experience: Mapped[int] = mapped_column(Integer, nullable=False)
group: Mapped[str] = mapped_column(String(50))
"""
@classmethod
def from_model(cls, model: User) -> Self:
"""
Create a user entity from model
Args: model (User): the model to create the entity from
Returns:
self: The entity
"""
return cls(
id=model.id,
username=model.username,
role=model.role,
email=model.email,
program=model.program,
experience=model.experience,
group=model.group,
program=model.programtype,
role=model.usertype,
created_at=model.created_at,
)
def to_model(self) -> User:
"""
Create a user model from entity
Returns:
User: A User model for API usage
"""
return User(
id=self.id,
username=self.id,
role=self.role,
email=self.email,
program=self.program,
experience=self.experience,
group=self.group,
username=self.username,
email= self.email ,
experience= self.experience,
group= self.group,
programtype= self.program,
usertype= self.role,
created_at= self.created_at,
)
"""

View File

@ -7,3 +7,7 @@ class TagService:
def __init__(self, session: Session = Depends(db_session)):
self._session = session
#get all tags - emma
def get_all_tags():
return

View File

@ -1,9 +1,72 @@
from fastapi import Depends
from ..database import db_session
from sqlalchemy.orm import Session
from ..entities.user_entity import UserEntity
from ..models.user_model import User
from sqlalchemy import select
class UserService:
def __init__(self, session: Session = Depends(db_session)):
self._session = session
def get_user_by_id(self) -> User:
"""
Gets a user by id from the database
Returns: A User Pydantic model
"""
user = (
self._session.query(UserEntity)
.filter(UserEntity.id == id)
)
if user is None:
raise Exception(
f"No user found with matching id: {id}"
)
return user.to_model()
#get users
def all(self) -> list[User]:
"""
Returns a list of all Users
"""
query = select(UserEntity)
entities = self._session.scalars(query).all()
return [entity.to_model() for entity in entities]
#post user
def create(self, user: User) -> User:
"""
Creates a new User Entity and adds to database
Args: User model
Returns: User model
"""
#handle if id exists
if user.id:
user.id = None
# if does not exist, create new object
user_entity = UserEntity.from_model(user)
# add new user to table
self._session.add(user_entity)
self._session.commit()
# return added object
return user_entity.to_model()

View File

@ -0,0 +1,55 @@
"""Tests for the UserService class."""
# PyTest
import pytest
from ...models.user_model import User
from ...services import UserService
from ...models.user_model import User
from ...entities.program_enum import ProgramEnum
from ...entities.user_enum import RoleEnum
from ...entities.user_entity import UserEntity
programs = ProgramEnum
roles = RoleEnum
volunteer = User(
id = 1,
username="volunteer",
email="volunteer@compass.com",
experience="1 year",
group="volunteers",
programtype=[programs.ECONOMIC],
usertype=roles.VOLUNTEER,
)
employee = User(
id = 2,
username="employee",
email="employee@compass.com",
experience="5 years",
group="employees",
programtype=[programs.DOMESTIC, programs.COMMUNITY],
usertype=roles.EMPLOYEE,
)
admin = User(
id = 3,
username="admin",
email="admin@compass.com",
experience="10 years",
group="admin",
programtype=[programs.DOMESTIC, programs.COMMUNITY, programs.ECONOMIC],
usertype=roles.ADMIN,
)
users=[volunteer, employee, admin]
def test_get_all():
"""Test that all users can be retrieved."""

View File

@ -0,0 +1,102 @@
import pytest
from sqlalchemy.orm import Session
from ...models.user_model import User
from ...entities.program_enum import ProgramEnum
from ...entities.user_enum import RoleEnum
from ...entities.user_entity import UserEntity
programs = ProgramEnum
roles = RoleEnum
volunteer = User(
id = 1,
username="volunteer",
email="volunteer@compass.com",
experience="1 year",
group="volunteers",
programtype=[programs.ECONOMIC],
usertype=roles.VOLUNTEER,
)
employee = User(
id = 2,
username="employee",
email="employee@compass.com",
experience="5 years",
group="employees",
programtype=[programs.DOMESTIC, programs.COMMUNITY],
usertype=roles.EMPLOYEE,
)
admin = User(
id = 3,
username="admin",
email="admin@compass.com",
experience="10 years",
group="admin",
programtype=[programs.DOMESTIC, programs.COMMUNITY, programs.ECONOMIC],
usertype=roles.ADMIN,
)
users=[volunteer, employee, admin]
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 organization data into the test session."""
global users
# Create entities for test organization data
entities = []
for user in users:
entity = UserEntity.from_model(user)
session.add(entity)
entities.append(entity)
# Reset table IDs to prevent ID conflicts
reset_table_id_seq(
session, UserEntity, UserEntity.id, len(users) + 1
)
# Commit all changes
session.commit()
@pytest.fixture(autouse=True)
def fake_data_fixture(session: Session):
"""Insert fake data the session automatically when test is run.
Note:
This function runs automatically due to the fixture property `autouse=True`.
"""
insert_fake_data(session)
session.commit()
yield