Compare commits
3 Commits
feature/hi
...
0a51e6559e
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0a51e6559e | ||
|
|
ab0d0a4323 | ||
|
|
076765e5c5 |
3
.idea/league-manager.iml
generated
3
.idea/league-manager.iml
generated
@@ -3,8 +3,9 @@
|
||||
<component name="NewModuleRootManager">
|
||||
<content url="file://$MODULE_DIR$">
|
||||
<sourceFolder url="file://$MODULE_DIR$/backend" isTestSource="false" />
|
||||
<excludeFolder url="file://$MODULE_DIR$/.venv" />
|
||||
</content>
|
||||
<orderEntry type="jdk" jdkName="Python 3.13" jdkType="Python SDK" />
|
||||
<orderEntry type="jdk" jdkName="Python 3.14 (RSW_puntensysteem)" jdkType="Python SDK" />
|
||||
<orderEntry type="sourceFolder" forTests="false" />
|
||||
</component>
|
||||
<component name="PyDocumentationSettings">
|
||||
|
||||
2
.idea/misc.xml
generated
2
.idea/misc.xml
generated
@@ -3,5 +3,5 @@
|
||||
<component name="Black">
|
||||
<option name="sdkName" value="Python 3.13" />
|
||||
</component>
|
||||
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.13" project-jdk-type="Python SDK" />
|
||||
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.14 (RSW_puntensysteem)" project-jdk-type="Python SDK" />
|
||||
</project>
|
||||
@@ -10,6 +10,7 @@ from app.api.routes import (
|
||||
private,
|
||||
users,
|
||||
utils,
|
||||
series,
|
||||
)
|
||||
from app.core.config import settings
|
||||
|
||||
@@ -24,6 +25,7 @@ api_router.include_router(teams.router)
|
||||
api_router.include_router(associations.router)
|
||||
api_router.include_router(divisions.router)
|
||||
api_router.include_router(members.router)
|
||||
api_router.include_router(series.router)
|
||||
|
||||
|
||||
if settings.ENVIRONMENT == "local":
|
||||
|
||||
176
backend/app/api/routes/series.py
Normal file
176
backend/app/api/routes/series.py
Normal file
@@ -0,0 +1,176 @@
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, HTTPException, status
|
||||
from sqlmodel import func, select
|
||||
|
||||
from app.api.deps import CurrentUser, SessionDep
|
||||
from app.models.base import (
|
||||
ApiTags,
|
||||
Message,
|
||||
RowId,
|
||||
)
|
||||
from app.models.serie import (
|
||||
Serie,
|
||||
SerieCreate,
|
||||
SerieUpdate,
|
||||
SeriePublic,
|
||||
SeriesPublic,
|
||||
)
|
||||
from app.models.division import (
|
||||
Division,
|
||||
DivisionsPublic,
|
||||
)
|
||||
from app.models.user import (
|
||||
PermissionModule,
|
||||
PermissionPart,
|
||||
PermissionRight,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/series", tags=[ApiTags.SERIES])
|
||||
|
||||
|
||||
# region # Series ########################################################
|
||||
|
||||
@router.get("/", response_model=SeriesPublic)
|
||||
def read_series(
|
||||
session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
|
||||
) -> Any:
|
||||
"""
|
||||
Retrieve all series.
|
||||
"""
|
||||
|
||||
if current_user.has_permissions(
|
||||
module=PermissionModule.SERIE,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
count_statement = select(func.count()).select_from(Serie)
|
||||
count = session.exec(count_statement).one()
|
||||
statement = select(Serie).offset(skip).limit(limit)
|
||||
series = session.exec(statement).all()
|
||||
return SeriesPublic(data=series, count=count)
|
||||
|
||||
return SeriesPublic(data=[], count=0)
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=SeriePublic)
|
||||
def read_serie(session: SessionDep, current_user: CurrentUser, id: RowId) -> Any:
|
||||
"""
|
||||
Get serie by ID.
|
||||
"""
|
||||
serie = session.get(Serie, id)
|
||||
if not serie:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Serie not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.SERIE,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
return serie
|
||||
|
||||
|
||||
@router.post("/", response_model=SeriePublic)
|
||||
def create_serie(
|
||||
*, session: SessionDep, current_user: CurrentUser, serie_in: SerieCreate
|
||||
) -> Any:
|
||||
"""
|
||||
Create new serie.
|
||||
"""
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.SERIE,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.CREATE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
serie = Serie.create(create_obj=serie_in, session=session)
|
||||
return serie
|
||||
|
||||
|
||||
@router.put("/{id}", response_model=SeriePublic)
|
||||
def update_serie(
|
||||
*, session: SessionDep, current_user: CurrentUser, id: RowId, serie_in: SerieUpdate
|
||||
) -> Any:
|
||||
"""
|
||||
Update a serie.
|
||||
"""
|
||||
serie = session.get(Serie, id)
|
||||
if not serie:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Serie not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.SERIE,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.UPDATE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
serie = Serie.update(db_obj=serie, in_obj=serie_in, session=session)
|
||||
return serie
|
||||
|
||||
|
||||
@router.delete("/{id}")
|
||||
def delete_serie(session: SessionDep,current_user: CurrentUser, id: RowId) -> Message:
|
||||
"""
|
||||
Delete a serie.
|
||||
"""
|
||||
serie = session.get(Serie, id)
|
||||
if not serie:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Serie not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.SERIE,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.DELETE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
session.delete(serie)
|
||||
session.commit()
|
||||
return Message(message="Serie deleted successfully")
|
||||
|
||||
# endregion
|
||||
|
||||
|
||||
# region # Series / Divisions ############################################
|
||||
|
||||
|
||||
@router.get("/{series_id}/divisions/", response_model=DivisionsPublic)
|
||||
def read_serie_division(
|
||||
session: SessionDep, current_user: CurrentUser, series_id: RowId, skip: int = 0, limit: int = 100
|
||||
) -> Any:
|
||||
"""
|
||||
Retrieve all serie divisions.
|
||||
"""
|
||||
|
||||
serie = session.get(Serie, series_id)
|
||||
if not serie:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Serie not found")
|
||||
|
||||
if not current_user.has_permission(
|
||||
module=PermissionModule.SERIE,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=(PermissionRight.MANAGE_DIVISIONS | PermissionRight.READ),
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
count_statement = (select(func.count())
|
||||
.select_from(Division)
|
||||
.where(Division.serie_id == serie.id)
|
||||
)
|
||||
count = session.exec(count_statement).one()
|
||||
statement = (select(Division)
|
||||
.where(Division.serie_id == serie.id)
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
)
|
||||
divisions = session.exec(statement).all()
|
||||
|
||||
return DivisionsPublic(data=divisions, count=count)
|
||||
|
||||
|
||||
# endregion
|
||||
@@ -30,6 +30,9 @@ from app.models.member import (
|
||||
from app.models.apikey import (
|
||||
ApiKey,
|
||||
)
|
||||
from app.models.serie import (
|
||||
Serie,
|
||||
)
|
||||
|
||||
engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))
|
||||
|
||||
|
||||
@@ -59,8 +59,14 @@ class ApiTags(DocumentedStrEnum):
|
||||
ASSOCIATIONS = "Associations"
|
||||
DIVISIONS = "Divisions"
|
||||
MEMBERS = "Members"
|
||||
SERIES = "Series"
|
||||
|
||||
|
||||
class VisitedCountType(DocumentedStrEnum):
|
||||
NONE = auto_enum()
|
||||
ONE_VISIT = auto_enum()
|
||||
EACH_VISIT = auto_enum()
|
||||
|
||||
# endregion
|
||||
|
||||
|
||||
|
||||
@@ -6,7 +6,7 @@ from sqlmodel import (
|
||||
Field,
|
||||
)
|
||||
|
||||
from .base import RowId as RowIdType
|
||||
from .base import RowId as RowIdType, VisitedCountType
|
||||
from ..core.config import settings
|
||||
|
||||
|
||||
@@ -120,3 +120,28 @@ class Birthday(BaseModel):
|
||||
class Created(BaseModel):
|
||||
created_at: datetime | None = Field(nullable=False, default_factory=lambda: datetime.now(settings.tz_info))
|
||||
created_by: RowIdType | None = Field(default=None, nullable=True, foreign_key="user.id", ondelete="SET NULL")
|
||||
|
||||
|
||||
class QuestionInfo(BaseModel):
|
||||
question_file: str | None = Field(default=None, nullable=True, max_length=255)
|
||||
answer_file: str | None = Field(default=None, nullable=True, max_length=255)
|
||||
|
||||
|
||||
class TeamAmmounts(BaseModel):
|
||||
min_teams: int = Field(default=None, nullable=True, ge=1)
|
||||
max_teams: int = Field(default=None, nullable=True, ge=1)
|
||||
|
||||
|
||||
class MaxPoints(BaseModel):
|
||||
max_points: int = Field(default=None, nullable=True, ge=0)
|
||||
|
||||
|
||||
class VisitedPoints(BaseModel):
|
||||
visited_points: int | None = Field(
|
||||
default=None,
|
||||
ge=0,
|
||||
description="Visited points for this place, None will disable this function",
|
||||
)
|
||||
visited_count_type: VisitedCountType = Field(
|
||||
default=VisitedCountType.NONE,
|
||||
)
|
||||
|
||||
206
backend/app/models/post.py
Normal file
206
backend/app/models/post.py
Normal file
@@ -0,0 +1,206 @@
|
||||
# app/models/post.py
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, select
|
||||
|
||||
from . import mixin
|
||||
from .base import BaseSQLModel, RowId, DocumentedStrEnum, DocumentedIntFlag, auto_enum
|
||||
from .user import PermissionRight, User
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .team import Team
|
||||
|
||||
# region # Post ##############################################################
|
||||
|
||||
class PostType(DocumentedIntFlag):
|
||||
POINTS = auto_enum() #allows for the ability to get points
|
||||
OPTION = auto_enum() #allows groups to enter waiting list
|
||||
POINTS_OPTIONS = POINTS | POINTS
|
||||
|
||||
VISITED = auto_enum() #only mark if a team has visited this post
|
||||
CORRECT = auto_enum() #possible to fill in if correct (get max points)
|
||||
|
||||
|
||||
class ReplayType(DocumentedStrEnum):
|
||||
NOT_POSSIBLE = auto_enum()
|
||||
ONLY_WHEN_NO_POINTS = auto_enum()
|
||||
YES_BUT_FIRST_POINTS_COUNT = auto_enum()
|
||||
YES_BUT_LAST_POINTS_COUNT = auto_enum()
|
||||
YES_BUT_HIGHEST_POINTS_COUNT = auto_enum()
|
||||
ALWAYS_AS_NEW = auto_enum() # add points together
|
||||
|
||||
# ##############################################################################
|
||||
|
||||
# Shared properties
|
||||
class PostUserLinkBase(BaseSQLModel):
|
||||
rights: PermissionRight = Field(default=PermissionRight.READ, nullable=False)
|
||||
|
||||
# Properties to receive via API on creation
|
||||
class PostUserLinkCreate(PostUserLinkBase):
|
||||
user_id: RowId = Field(nullable=False)
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class PostUserLinkUpdate(PostUserLinkBase):
|
||||
pass
|
||||
|
||||
# Database model (link tussen post en user)
|
||||
class PostUserLink(PostUserLinkBase, table=True):
|
||||
post_id: RowId = Field(
|
||||
foreign_key="post.id",
|
||||
primary_key=True,
|
||||
nullable=False,
|
||||
ondelete="CASCADE",
|
||||
)
|
||||
user_id: RowId = Field(
|
||||
foreign_key="user.id",
|
||||
primary_key=True,
|
||||
nullable=False,
|
||||
ondelete="CASCADE",
|
||||
)
|
||||
|
||||
post: "Post" = Relationship(back_populates="user_links")
|
||||
user: "User" = Relationship(back_populates="post_links")
|
||||
|
||||
# Properties to return via API
|
||||
class PostUserLinkPublic(PostUserLinkBase):
|
||||
user_id: RowId
|
||||
post_id: RowId
|
||||
|
||||
class PostUserLinksPublic(BaseSQLModel):
|
||||
data: list[PostUserLinkPublic]
|
||||
count: int
|
||||
|
||||
# Shared properties
|
||||
class PostBase(
|
||||
#common
|
||||
mixin.Name, #post name
|
||||
mixin.Contact, #name contact person
|
||||
mixin.IsActive, #scouting active (unused)
|
||||
|
||||
#post specific
|
||||
mixin.ShortName, #shortname for post
|
||||
|
||||
mixin.MaxPoints, #max obtainable points
|
||||
mixin.VisitedPoints, #points obtained for visiting post
|
||||
|
||||
mixin.TeamAmmounts, #teams, min&max teams required
|
||||
|
||||
mixin.QuestionInfo, #field for questions
|
||||
|
||||
BaseSQLModel,
|
||||
):
|
||||
post_type: PostType = Field(default=PostType.POINTS, nullable=False)
|
||||
replay: ReplayType = Field(default=ReplayType.NOT_POSSIBLE, nullable=False)
|
||||
|
||||
# Properties to receive via API on creation
|
||||
class PostCreate(PostBase):
|
||||
pass
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class PostUpdate(mixin.ShortNameUpdate, PostBase):
|
||||
post_type: PostType | None = Field(default=None, nullable=True) # type: ignore
|
||||
replay: ReplayType | None = Field(default=None, nullable=True) # type: ignore
|
||||
|
||||
# Database model
|
||||
class Post(mixin.RowId, PostBase, table=True):
|
||||
|
||||
user_links: list["PostUserLink"] = Relationship(back_populates="post", cascade_delete=True)
|
||||
#teams: list["Team"] = Relationship(back_populates="post", cascade_delete=True)
|
||||
|
||||
@classmethod
|
||||
def create(cls, *, session: Session, create_obj: PostCreate) -> "Post":
|
||||
data_obj = create_obj.model_dump(exclude_unset=True)
|
||||
db_obj = cls.model_validate(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
@classmethod
|
||||
def update(cls, *, session: Session, db_obj: "Post", in_obj: PostUpdate) -> "Post":
|
||||
data_obj = in_obj.model_dump(exclude_unset=True)
|
||||
db_obj.sqlmodel_update(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def get_user_link(self, user: User) -> "PostUserLink | None":
|
||||
return next((link for link in self.user_links if link.user == user), None)
|
||||
|
||||
def add_user(
|
||||
self,
|
||||
user: User,
|
||||
rights: PermissionRight = PermissionRight.READ,
|
||||
*,
|
||||
session: Session,
|
||||
) -> "PostUserLink | None":
|
||||
existing = self.get_user_link(user=user)
|
||||
if existing is None:
|
||||
new_link = PostUserLink(post=self, user=user, rights=rights)
|
||||
self.user_links.append(new_link)
|
||||
session.add(new_link)
|
||||
session.commit()
|
||||
return new_link
|
||||
return None
|
||||
|
||||
def update_user(
|
||||
self,
|
||||
user: User,
|
||||
rights: PermissionRight = PermissionRight.READ,
|
||||
*,
|
||||
session: Session,
|
||||
) -> "PostUserLink | None":
|
||||
link = self.get_user_link(user=user)
|
||||
if link:
|
||||
link.rights = rights
|
||||
session.add(link)
|
||||
session.commit()
|
||||
return link
|
||||
return None
|
||||
|
||||
def remove_user(self, user: User, *, session: Session) -> None:
|
||||
link = self.get_user_link(user=user)
|
||||
if link:
|
||||
session.delete(link)
|
||||
session.commit()
|
||||
|
||||
def user_has_rights(
|
||||
self,
|
||||
user: User,
|
||||
rights: PermissionRight | None = None,
|
||||
) -> bool:
|
||||
return any(
|
||||
(
|
||||
link.user == user
|
||||
and link.rights
|
||||
and (not rights or (link.rights & rights) == rights)
|
||||
)
|
||||
for link in self.user_links
|
||||
)
|
||||
|
||||
def user_has_right(
|
||||
self,
|
||||
user: User,
|
||||
rights: PermissionRight | None = None,
|
||||
) -> bool:
|
||||
return any(
|
||||
(
|
||||
link.user == user
|
||||
and link.rights
|
||||
and (not rights or (link.rights & rights))
|
||||
)
|
||||
for link in self.user_links
|
||||
)
|
||||
|
||||
|
||||
# API output models
|
||||
class PostPublic(mixin.RowIdPublic, PostBase):
|
||||
pass
|
||||
|
||||
|
||||
class PostsPublic(BaseSQLModel):
|
||||
data: list[PostPublic]
|
||||
count: int
|
||||
|
||||
# endregion
|
||||
173
backend/app/models/serie.py
Normal file
173
backend/app/models/serie.py
Normal file
@@ -0,0 +1,173 @@
|
||||
# app/models/serie.py
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlmodel import Field, Relationship, Session, select
|
||||
|
||||
from . import mixin
|
||||
from .base import BaseSQLModel, RowId
|
||||
from .user import PermissionRight, User
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .team import Team
|
||||
|
||||
# region # Serie ##############################################################
|
||||
|
||||
# Shared properties
|
||||
class SerieUserLinkBase(BaseSQLModel):
|
||||
rights: PermissionRight = Field(default=PermissionRight.READ, nullable=False)
|
||||
|
||||
# Properties to receive via API on creation
|
||||
class SerieUserLinkCreate(SerieUserLinkBase):
|
||||
user_id: RowId = Field(nullable=False)
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class SerieUserLinkUpdate(SerieUserLinkBase):
|
||||
pass
|
||||
|
||||
# Database model (link tussen serie en user)
|
||||
class SerieUserLink(SerieUserLinkBase, table=True):
|
||||
serie_id: RowId = Field(
|
||||
foreign_key="serie.id",
|
||||
primary_key=True,
|
||||
nullable=False,
|
||||
ondelete="CASCADE",
|
||||
)
|
||||
user_id: RowId = Field(
|
||||
foreign_key="user.id",
|
||||
primary_key=True,
|
||||
nullable=False,
|
||||
ondelete="CASCADE",
|
||||
)
|
||||
|
||||
serie: "Serie" = Relationship(back_populates="user_links")
|
||||
user: "User" = Relationship(back_populates="serie_links")
|
||||
|
||||
# Properties to return via API
|
||||
class SerieUserLinkPublic(SerieUserLinkBase):
|
||||
user_id: RowId
|
||||
serie_id: RowId
|
||||
|
||||
class SerieUserLinksPublic(BaseSQLModel):
|
||||
data: list[SerieUserLinkPublic]
|
||||
count: int
|
||||
|
||||
# Shared properties
|
||||
class SerieBase(
|
||||
mixin.Name,
|
||||
mixin.Contact,
|
||||
mixin.IsActive,
|
||||
BaseSQLModel,
|
||||
):
|
||||
pass
|
||||
|
||||
# Properties to receive via API on creation
|
||||
class SerieCreate(SerieBase):
|
||||
pass
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class SerieUpdate(SerieBase):
|
||||
pass
|
||||
|
||||
# Database model
|
||||
class Serie(mixin.RowId, SerieBase, table=True):
|
||||
|
||||
user_links: list["SerieUserLink"] = Relationship(back_populates="serie", cascade_delete=True)
|
||||
#teams: list["Team"] = Relationship(back_populates="serie", cascade_delete=True)
|
||||
|
||||
@classmethod
|
||||
def create(cls, *, session: Session, create_obj: SerieCreate) -> "Serie":
|
||||
data_obj = create_obj.model_dump(exclude_unset=True)
|
||||
db_obj = cls.model_validate(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
@classmethod
|
||||
def update(cls, *, session: Session, db_obj: "Serie", in_obj: SerieUpdate) -> "Serie":
|
||||
data_obj = in_obj.model_dump(exclude_unset=True)
|
||||
db_obj.sqlmodel_update(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def get_user_link(self, user: User) -> "SerieUserLink | None":
|
||||
return next((link for link in self.user_links if link.user == user), None)
|
||||
|
||||
def add_user(
|
||||
self,
|
||||
user: User,
|
||||
rights: PermissionRight = PermissionRight.READ,
|
||||
*,
|
||||
session: Session,
|
||||
) -> "SerieUserLink | None":
|
||||
existing = self.get_user_link(user=user)
|
||||
if existing is None:
|
||||
new_link = SerieUserLink(serie=self, user=user, rights=rights)
|
||||
self.user_links.append(new_link)
|
||||
session.add(new_link)
|
||||
session.commit()
|
||||
return new_link
|
||||
return None #TODO aanpassen??
|
||||
|
||||
def update_user(
|
||||
self,
|
||||
user: User,
|
||||
rights: PermissionRight = PermissionRight.READ,
|
||||
*,
|
||||
session: Session,
|
||||
) -> "SerieUserLink | None":
|
||||
link = self.get_user_link(user=user)
|
||||
if link:
|
||||
link.rights = rights
|
||||
session.add(link)
|
||||
session.commit()
|
||||
return link
|
||||
return None
|
||||
|
||||
def remove_user(self, user: User, *, session: Session) -> None:
|
||||
link = self.get_user_link(user=user)
|
||||
if link:
|
||||
session.delete(link)
|
||||
session.commit()
|
||||
|
||||
def user_has_rights(
|
||||
self,
|
||||
user: User,
|
||||
rights: PermissionRight | None = None,
|
||||
) -> bool:
|
||||
return any(
|
||||
(
|
||||
link.user == user
|
||||
and link.rights
|
||||
and (not rights or (link.rights & rights) == rights)
|
||||
)
|
||||
for link in self.user_links
|
||||
)
|
||||
|
||||
def user_has_right(
|
||||
self,
|
||||
user: User,
|
||||
rights: PermissionRight | None = None,
|
||||
) -> bool:
|
||||
return any(
|
||||
(
|
||||
link.user == user
|
||||
and link.rights
|
||||
and (not rights or (link.rights & rights))
|
||||
)
|
||||
for link in self.user_links
|
||||
)
|
||||
|
||||
|
||||
# API output models
|
||||
class SeriePublic(mixin.RowIdPublic, SerieBase):
|
||||
pass
|
||||
|
||||
|
||||
class SeriesPublic(BaseSQLModel):
|
||||
data: list[SeriePublic]
|
||||
count: int
|
||||
|
||||
# endregion
|
||||
@@ -18,6 +18,7 @@ if TYPE_CHECKING:
|
||||
from .apikey import ApiKey
|
||||
from .event import EventUserLink
|
||||
from .member import Member
|
||||
from .serie import SerieUserLink
|
||||
|
||||
|
||||
# region # User ################################################################
|
||||
@@ -31,6 +32,8 @@ class PermissionModule(DocumentedStrEnum):
|
||||
ASSOCIATION = auto_enum()
|
||||
DIVISION = auto_enum()
|
||||
MEMBER = auto_enum()
|
||||
SERIE = auto_enum()
|
||||
POST = auto_enum()
|
||||
|
||||
|
||||
class PermissionPart(DocumentedStrEnum):
|
||||
@@ -136,6 +139,7 @@ class User(mixin.RowId, UserBase, table=True):
|
||||
# --- many-to-many links ---------------------------------------------------
|
||||
roles: list["Role"] = Relationship(back_populates="users", link_model=UserRoleLink)
|
||||
event_links: list["EventUserLink"] = Relationship(back_populates="user")
|
||||
serie_links: list["SerieUserLink"] = Relationship(back_populates="user")
|
||||
|
||||
# --- CRUD actions ---------------------------------------------------------
|
||||
@classmethod
|
||||
|
||||
179
backend/app/tests/api/routes/test_serie.py
Normal file
179
backend/app/tests/api/routes/test_serie.py
Normal file
@@ -0,0 +1,179 @@
|
||||
import uuid
|
||||
|
||||
from fastapi import status
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.core.config import settings
|
||||
from app.tests.utils.serie import create_random_serie
|
||||
|
||||
|
||||
def test_create_serie(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
data = {
|
||||
"name": "Zondag spel",
|
||||
"contact": "Rick",
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/series/",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == data["name"]
|
||||
assert content["contact"] == data["contact"]
|
||||
assert "id" in content
|
||||
|
||||
|
||||
def test_create_serie_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
data = {
|
||||
"name": "Zondag Spel",
|
||||
"contact": "Rick",
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/series/",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_serie(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
serie = create_random_serie(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/series/{serie.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(serie.id)
|
||||
assert content["name"] == serie.name
|
||||
assert content["contact"] == serie.contact
|
||||
|
||||
|
||||
def test_read_serie_not_found(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/series/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Serie not found"
|
||||
|
||||
|
||||
def test_read_serie_no_permission(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
serie = create_random_serie(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/series/{serie.id}",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_series(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
create_random_serie(db)
|
||||
create_random_serie(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/series/",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert "count" in content
|
||||
assert content["count"] >= 2
|
||||
assert "data" in content
|
||||
assert isinstance(content["data"], list)
|
||||
assert len(content["data"]) <= content["count"]
|
||||
|
||||
|
||||
def test_read_series_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
create_random_serie(db)
|
||||
create_random_serie(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/series/",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert "count" in content
|
||||
assert content["count"] == 0
|
||||
assert "data" in content
|
||||
assert isinstance(content["data"], list)
|
||||
assert len(content["data"]) == 0
|
||||
|
||||
|
||||
def test_update_serie(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
serie = create_random_serie(db)
|
||||
data = {
|
||||
"name": "Updated name",
|
||||
"contact": "Updated contact",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/series/{serie.id}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(serie.id)
|
||||
assert content["name"] == data["name"]
|
||||
assert content["contact"] == data["contact"]
|
||||
|
||||
|
||||
def test_update_serie_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
data = {
|
||||
"name": "Not found",
|
||||
"contact": "Not found",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/series/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Serie not found"
|
||||
|
||||
|
||||
def test_update_serie_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
serie = create_random_serie(db)
|
||||
data = {
|
||||
"name": "No permissions",
|
||||
"contact": "No permissions",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/series/{serie.id}",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_delete_serie(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
serie = create_random_serie(db)
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/series/{serie.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["message"] == "Serie deleted successfully"
|
||||
|
||||
|
||||
def test_delete_serie_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/series/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Serie not found"
|
||||
|
||||
|
||||
def test_delete_serie_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
serie = create_random_serie(db)
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/series/{serie.id}",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
12
backend/app/tests/utils/serie.py
Normal file
12
backend/app/tests/utils/serie.py
Normal file
@@ -0,0 +1,12 @@
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.models.serie import Serie, SerieCreate
|
||||
from app.tests.utils.utils import random_lower_string
|
||||
|
||||
|
||||
def create_random_serie(db: Session, name: str = None) -> Serie:
|
||||
if not name:
|
||||
name = random_lower_string()
|
||||
|
||||
serie_in = SerieCreate(name=name)
|
||||
return Serie.create(session=db, create_obj=serie_in)
|
||||
Reference in New Issue
Block a user