Refactoring the history tab of the requirements details

This commit is contained in:
gulimabr
2025-12-03 16:41:19 -03:00
parent fef1545cdb
commit 9298c80eb2
9 changed files with 916 additions and 80 deletions

View File

@@ -286,6 +286,11 @@ class RequirementGroup(Base):
ForeignKey("groups.id", ondelete="CASCADE"),
primary_key=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=datetime.utcnow,
nullable=True
)
class Validation(Base):
@@ -328,6 +333,7 @@ class RequirementHistory(Base):
history_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
original_req_id: Mapped[int] = mapped_column(Integer, nullable=False)
project_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
status_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) # Requirement lifecycle status
req_name: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
req_desc: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
priority_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
@@ -512,3 +518,29 @@ class RequirementCommentReply(Base):
__table_args__ = (
Index("idx_rcr_parent", "parent_comment_id"),
)
class RequirementLinkHistory(Base):
"""
Historical records of requirement link changes (deletions).
Note: This is populated by a database trigger on UPDATE/DELETE.
Stores snapshots of relationship type names to preserve data if types are renamed/deleted.
"""
__tablename__ = "requirement_links_history"
history_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
original_link_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
source_req_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
target_req_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
relationship_type_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
relationship_type_snapshot: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # Preserved type name
inverse_type_snapshot: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # Preserved inverse name
created_by: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
valid_from: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) # When link was created
valid_to: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) # When link was deleted
# Indexes
__table_args__ = (
Index("idx_link_hist_source", "source_req_id"),
Index("idx_link_hist_target", "target_req_id"),
)

View File

@@ -13,6 +13,7 @@ from src.models import (
ValidationStatusResponse, ValidationHistoryResponse, ValidationCreateRequest,
RelationshipTypeResponse, RelationshipTypeCreateRequest, RelationshipTypeUpdateRequest,
RequirementLinkResponse, RequirementLinkCreateRequest, RequirementSearchResult,
RequirementLinkHistoryResponse, RequirementGroupHistoryResponse, CurrentRequirementGroupResponse,
RoleResponse, ProjectMemberResponse, UserRoleUpdateRequest, ROLE_DISPLAY_NAMES,
CommentResponse, CommentReplyResponse, CommentCreateRequest, ReplyCreateRequest,
RequirementStatusResponse, DeletedRequirementResponse
@@ -1600,6 +1601,117 @@ async def delete_requirement_link(
await db.commit()
@app.get("/api/requirements/{requirement_id}/links/history", response_model=List[RequirementLinkHistoryResponse])
async def get_requirement_link_history(
requirement_id: int,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""
Get the link history for a requirement (deleted/changed links).
Returns all historical links where this requirement was source or target.
Args:
requirement_id: The requirement to get link history for
Returns:
List of historical links with relationship type snapshots and requirement info.
"""
user = await _get_current_user_db(request, db)
# Check if requirement exists
req_repo = RequirementRepository(db)
requirement = await req_repo.get_by_id(requirement_id)
if not requirement:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Requirement with id {requirement_id} not found"
)
# Verify user is a member of the requirement's project
await _verify_project_membership(requirement.project_id, user.id, db)
# Get link history
link_repo = RequirementLinkRepository(db)
history = await link_repo.get_history_by_requirement_id(requirement_id)
return [RequirementLinkHistoryResponse(**h) for h in history]
@app.get("/api/requirements/{requirement_id}/groups/history", response_model=List[RequirementGroupHistoryResponse])
async def get_requirement_group_history(
requirement_id: int,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""
Get the group association history for a requirement (removed groups).
Returns all historical group associations for this requirement.
Args:
requirement_id: The requirement to get group history for
Returns:
List of historical group associations with group name/color snapshots.
"""
user = await _get_current_user_db(request, db)
# Check if requirement exists
req_repo = RequirementRepository(db)
requirement = await req_repo.get_by_id(requirement_id)
if not requirement:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Requirement with id {requirement_id} not found"
)
# Verify user is a member of the requirement's project
await _verify_project_membership(requirement.project_id, user.id, db)
# Get group history
group_repo = GroupRepository(db)
history = await group_repo.get_group_history_by_requirement_id(requirement_id)
return [RequirementGroupHistoryResponse(**h) for h in history]
@app.get("/api/requirements/{requirement_id}/groups/current", response_model=List[CurrentRequirementGroupResponse])
async def get_requirement_current_groups(
requirement_id: int,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""
Get the current groups for a requirement with their association timestamps.
Returns all current group associations with when they were added.
Args:
requirement_id: The requirement to get current groups for
Returns:
List of current group associations with group name/color and created_at timestamp.
"""
user = await _get_current_user_db(request, db)
# Check if requirement exists
req_repo = RequirementRepository(db)
requirement = await req_repo.get_by_id(requirement_id)
if not requirement:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Requirement with id {requirement_id} not found"
)
# Verify user is a member of the requirement's project
await _verify_project_membership(requirement.project_id, user.id, db)
# Get current groups
group_repo = GroupRepository(db)
groups = await group_repo.get_current_groups_by_requirement_id(requirement_id)
return [CurrentRequirementGroupResponse(**g) for g in groups]
# ===========================================
# Comment Endpoints
# ===========================================

View File

@@ -351,6 +351,26 @@ class RequirementLinkCreateRequest(BaseModel):
target_requirement_id: int
class RequirementLinkHistoryResponse(BaseModel):
"""Response schema for requirement link history (deleted/changed links)."""
history_id: int
original_link_id: Optional[int] = None
source_req_id: Optional[int] = None
target_req_id: Optional[int] = None
source_tag_code: Optional[str] = None # Resolved from source requirement
target_tag_code: Optional[str] = None # Resolved from target requirement
source_req_name: Optional[str] = None
target_req_name: Optional[str] = None
relationship_type_snapshot: Optional[str] = None # Preserved type name
inverse_type_snapshot: Optional[str] = None # Preserved inverse name
created_by_username: Optional[str] = None
valid_from: Optional[datetime] = None # When link was created
valid_to: Optional[datetime] = None # When link was deleted
class Config:
from_attributes = True
# Requirement Search schemas
class RequirementSearchResult(BaseModel):
"""Response schema for requirement search results (for autocomplete)."""
@@ -402,3 +422,29 @@ class CommentCreateRequest(BaseModel):
class ReplyCreateRequest(BaseModel):
"""Request schema for creating a reply."""
reply_text: str
class RequirementGroupHistoryResponse(BaseModel):
"""Response schema for requirement group association history (removed groups)."""
history_id: int
requirement_id: int
group_id: int
group_name_snapshot: Optional[str] = None # Preserved group name at time of removal
group_hex_color_snapshot: Optional[str] = None # Preserved hex color at time of removal
valid_from: Optional[datetime] = None # When the group was associated
valid_to: Optional[datetime] = None # When the group was removed
class Config:
from_attributes = True
class CurrentRequirementGroupResponse(BaseModel):
"""Response schema for current group associations with timestamp."""
requirement_id: int
group_id: int
group_name: str
group_hex_color: str
created_at: Optional[datetime] = None # When the group was added
class Config:
from_attributes = True

View File

@@ -1,8 +1,8 @@
"""
Repository layer for Group database operations.
"""
from typing import List, Optional
from sqlalchemy import select
from typing import List, Optional, Dict, Any
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession
from src.db_models import Group
import logging
@@ -74,3 +74,81 @@ class GroupRepository:
await self.session.flush()
await self.session.refresh(group)
return group
async def get_group_history_by_requirement_id(self, requirement_id: int) -> List[Dict[str, Any]]:
"""
Get the group association history for a requirement.
Returns records of when groups were added or removed from this requirement.
Args:
requirement_id: The requirement ID to get group history for
Returns:
List of historical group associations with snapshots of group data
"""
query = text("""
SELECT
rgh.history_id,
rgh.requirement_id,
rgh.group_id,
rgh.group_name_snapshot,
rgh.group_hex_color_snapshot,
rgh.valid_from,
rgh.valid_to
FROM requirements_groups_history rgh
WHERE rgh.requirement_id = :requirement_id
ORDER BY rgh.valid_to DESC
""")
result = await self.session.execute(query, {"requirement_id": requirement_id})
rows = result.fetchall()
return [
{
"history_id": row.history_id,
"requirement_id": row.requirement_id,
"group_id": row.group_id,
"group_name_snapshot": row.group_name_snapshot,
"group_hex_color_snapshot": row.group_hex_color_snapshot,
"valid_from": row.valid_from,
"valid_to": row.valid_to,
}
for row in rows
]
async def get_current_groups_by_requirement_id(self, requirement_id: int) -> List[Dict[str, Any]]:
"""
Get the current groups associated with a requirement, including when they were added.
Args:
requirement_id: The requirement ID to get groups for
Returns:
List of current group associations with group data and created_at timestamp
"""
query = text("""
SELECT
rg.requirement_id,
rg.group_id,
g.group_name,
g.hex_color,
rg.created_at
FROM requirements_groups rg
JOIN groups g ON rg.group_id = g.id
WHERE rg.requirement_id = :requirement_id
ORDER BY rg.created_at DESC
""")
result = await self.session.execute(query, {"requirement_id": requirement_id})
rows = result.fetchall()
return [
{
"requirement_id": row.requirement_id,
"group_id": row.group_id,
"group_name": row.group_name,
"group_hex_color": row.hex_color,
"created_at": row.created_at,
}
for row in rows
]

View File

@@ -2,7 +2,7 @@
Repository for RequirementLink database operations.
"""
from typing import List, Optional, Dict, Any
from sqlalchemy import select, or_
from sqlalchemy import select, or_, text
from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import AsyncSession
from src.db_models import RequirementLink, Requirement, RelationshipType, User
@@ -197,3 +197,65 @@ class RequirementLinkRepository:
)
)
return result.scalar_one_or_none() is not None
async def get_history_by_requirement_id(self, requirement_id: int) -> List[Dict[str, Any]]:
"""
Get the link history for a requirement (both as source and target).
This data is populated by a database trigger on UPDATE/DELETE.
Args:
requirement_id: The requirement ID to get link history for
Returns:
List of historical links with resolved requirement names and creator info
"""
# Use raw SQL with LEFT JOINs to resolve foreign keys to display names
# Note: source/target requirements may have been deleted, so we use LEFT JOINs
query = text("""
SELECT
lh.history_id,
lh.original_link_id,
lh.source_req_id,
lh.target_req_id,
lh.relationship_type_snapshot,
lh.inverse_type_snapshot,
lh.valid_from,
lh.valid_to,
u.full_name as created_by_full_name,
u.username as created_by_username,
sr.req_name as source_req_name,
st.tag_code as source_tag_code,
tr.req_name as target_req_name,
tt.tag_code as target_tag_code
FROM requirement_links_history lh
LEFT JOIN users u ON lh.created_by = u.id
LEFT JOIN requirements sr ON lh.source_req_id = sr.id
LEFT JOIN tags st ON sr.tag_id = st.id
LEFT JOIN requirements tr ON lh.target_req_id = tr.id
LEFT JOIN tags tt ON tr.tag_id = tt.id
WHERE lh.source_req_id = :requirement_id
OR lh.target_req_id = :requirement_id
ORDER BY lh.valid_to DESC
""")
result = await self.db.execute(query, {"requirement_id": requirement_id})
rows = result.fetchall()
return [
{
"history_id": row.history_id,
"original_link_id": row.original_link_id,
"source_req_id": row.source_req_id,
"target_req_id": row.target_req_id,
"source_tag_code": row.source_tag_code,
"target_tag_code": row.target_tag_code,
"source_req_name": row.source_req_name,
"target_req_name": row.target_req_name,
"relationship_type_snapshot": row.relationship_type_snapshot,
"inverse_type_snapshot": row.inverse_type_snapshot,
"created_by_username": row.created_by_full_name or row.created_by_username,
"valid_from": row.valid_from,
"valid_to": row.valid_to,
}
for row in rows
]