Added relationship between requirements
This commit is contained in:
gulimabr
@@ -57,6 +57,7 @@ class User(Base):
|
||||
secondary="project_members",
|
||||
back_populates="members"
|
||||
)
|
||||
created_links: Mapped[List["RequirementLink"]] = relationship("RequirementLink", back_populates="creator")
|
||||
|
||||
|
||||
class Tag(Base):
|
||||
@@ -119,6 +120,7 @@ class Project(Base):
|
||||
back_populates="projects"
|
||||
)
|
||||
requirements: Mapped[List["Requirement"]] = relationship("Requirement", back_populates="project")
|
||||
relationship_types: Mapped[List["RelationshipType"]] = relationship("RelationshipType", back_populates="project")
|
||||
|
||||
|
||||
class ProjectMember(Base):
|
||||
@@ -211,6 +213,16 @@ class Requirement(Base):
|
||||
back_populates="requirements"
|
||||
)
|
||||
validations: Mapped[List["Validation"]] = relationship("Validation", back_populates="requirement")
|
||||
outgoing_links: Mapped[List["RequirementLink"]] = relationship(
|
||||
"RequirementLink",
|
||||
foreign_keys="RequirementLink.source_req_id",
|
||||
back_populates="source_requirement"
|
||||
)
|
||||
incoming_links: Mapped[List["RequirementLink"]] = relationship(
|
||||
"RequirementLink",
|
||||
foreign_keys="RequirementLink.target_req_id",
|
||||
back_populates="target_requirement"
|
||||
)
|
||||
|
||||
# Indexes
|
||||
__table_args__ = (
|
||||
@@ -285,3 +297,85 @@ class RequirementHistory(Base):
|
||||
valid_from: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
|
||||
valid_to: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
|
||||
edited_by: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
|
||||
|
||||
|
||||
class RelationshipType(Base):
|
||||
"""
|
||||
Defines valid relationship types per project.
|
||||
E.g., "Depends On", "Parent", "Conflicts With".
|
||||
"""
|
||||
__tablename__ = "relationship_types"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
project_id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("projects.id", ondelete="CASCADE"),
|
||||
nullable=False
|
||||
)
|
||||
type_name: Mapped[str] = mapped_column(Text, nullable=False)
|
||||
type_description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
|
||||
inverse_type_name: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
|
||||
|
||||
# Relationships
|
||||
project: Mapped["Project"] = relationship("Project", back_populates="relationship_types")
|
||||
links: Mapped[List["RequirementLink"]] = relationship("RequirementLink", back_populates="relationship_type")
|
||||
|
||||
# Constraints
|
||||
__table_args__ = (
|
||||
UniqueConstraint("project_id", "type_name", name="uq_rel_type_name_project"),
|
||||
)
|
||||
|
||||
|
||||
class RequirementLink(Base):
|
||||
"""
|
||||
Links between requirements (e.g., dependencies, parent/child, conflicts).
|
||||
"""
|
||||
__tablename__ = "requirement_links"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
source_req_id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("requirements.id", ondelete="CASCADE"),
|
||||
nullable=False
|
||||
)
|
||||
target_req_id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("requirements.id", ondelete="CASCADE"),
|
||||
nullable=False
|
||||
)
|
||||
relationship_type_id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("relationship_types.id", ondelete="CASCADE"),
|
||||
nullable=False
|
||||
)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True),
|
||||
default=datetime.utcnow,
|
||||
nullable=True
|
||||
)
|
||||
created_by: Mapped[Optional[int]] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("users.id", ondelete="SET NULL"),
|
||||
nullable=True
|
||||
)
|
||||
|
||||
# Relationships
|
||||
source_requirement: Mapped["Requirement"] = relationship(
|
||||
"Requirement",
|
||||
foreign_keys=[source_req_id],
|
||||
back_populates="outgoing_links"
|
||||
)
|
||||
target_requirement: Mapped["Requirement"] = relationship(
|
||||
"Requirement",
|
||||
foreign_keys=[target_req_id],
|
||||
back_populates="incoming_links"
|
||||
)
|
||||
relationship_type: Mapped["RelationshipType"] = relationship("RelationshipType", back_populates="links")
|
||||
creator: Mapped[Optional["User"]] = relationship("User", back_populates="created_links")
|
||||
|
||||
# Indexes and constraints
|
||||
__table_args__ = (
|
||||
Index("idx_link_source", "source_req_id"),
|
||||
Index("idx_link_target", "target_req_id"),
|
||||
UniqueConstraint("source_req_id", "target_req_id", "relationship_type_id", name="uq_req_link_pair"),
|
||||
)
|
||||
|
||||
@@ -10,14 +10,17 @@ from src.models import (
|
||||
TagResponse, RequirementResponse, PriorityResponse,
|
||||
RequirementCreateRequest, RequirementUpdateRequest,
|
||||
ProjectResponse, ProjectCreateRequest, ProjectUpdateRequest, ProjectMemberRequest,
|
||||
ValidationStatusResponse, ValidationHistoryResponse, ValidationCreateRequest
|
||||
ValidationStatusResponse, ValidationHistoryResponse, ValidationCreateRequest,
|
||||
RelationshipTypeResponse, RelationshipTypeCreateRequest,
|
||||
RequirementLinkResponse, RequirementLinkCreateRequest, RequirementSearchResult
|
||||
)
|
||||
from src.controller import AuthController
|
||||
from src.config import get_openid, get_settings
|
||||
from src.database import init_db, close_db, get_db
|
||||
from src.repositories import (
|
||||
RoleRepository, GroupRepository, TagRepository, RequirementRepository,
|
||||
PriorityRepository, ProjectRepository, ValidationStatusRepository, ValidationRepository
|
||||
PriorityRepository, ProjectRepository, ValidationStatusRepository, ValidationRepository,
|
||||
RelationshipTypeRepository, RequirementLinkRepository
|
||||
)
|
||||
import logging
|
||||
|
||||
@@ -926,3 +929,297 @@ async def get_validation_history(
|
||||
)
|
||||
for v in validations
|
||||
]
|
||||
|
||||
|
||||
# ===========================================
|
||||
# Relationship Types Endpoints
|
||||
# ===========================================
|
||||
|
||||
@app.get("/api/projects/{project_id}/relationship-types", response_model=List[RelationshipTypeResponse])
|
||||
async def get_project_relationship_types(
|
||||
project_id: int,
|
||||
request: Request,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Get all relationship types for a project.
|
||||
User must be a member of the project.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
|
||||
Returns:
|
||||
List of relationship types for the project.
|
||||
"""
|
||||
user = await _get_current_user_db(request, db)
|
||||
await _verify_project_membership(project_id, user.id, db)
|
||||
|
||||
rel_type_repo = RelationshipTypeRepository(db)
|
||||
rel_types = await rel_type_repo.get_by_project_id(project_id)
|
||||
return [RelationshipTypeResponse.model_validate(rt) for rt in rel_types]
|
||||
|
||||
|
||||
@app.post("/api/projects/{project_id}/relationship-types", response_model=RelationshipTypeResponse, status_code=status.HTTP_201_CREATED)
|
||||
async def create_relationship_type(
|
||||
project_id: int,
|
||||
request: Request,
|
||||
type_data: RelationshipTypeCreateRequest,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Create a new relationship type for a project.
|
||||
Only admins (role_id=1) can create relationship types.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
type_data: The relationship type data
|
||||
|
||||
Returns:
|
||||
The created relationship type.
|
||||
"""
|
||||
user = await _get_current_user_db(request, db)
|
||||
|
||||
# Only admins can create relationship types
|
||||
_require_role(user, [1], "create relationship types")
|
||||
|
||||
await _verify_project_membership(project_id, user.id, db)
|
||||
|
||||
rel_type_repo = RelationshipTypeRepository(db)
|
||||
rel_type = await rel_type_repo.create(
|
||||
project_id=project_id,
|
||||
type_name=type_data.type_name,
|
||||
type_description=type_data.type_description,
|
||||
inverse_type_name=type_data.inverse_type_name
|
||||
)
|
||||
|
||||
await db.commit()
|
||||
return RelationshipTypeResponse.model_validate(rel_type)
|
||||
|
||||
|
||||
# ===========================================
|
||||
# Requirement Search Endpoint (for autocomplete)
|
||||
# ===========================================
|
||||
|
||||
@app.get("/api/projects/{project_id}/requirements/search", response_model=List[RequirementSearchResult])
|
||||
async def search_requirements(
|
||||
project_id: int,
|
||||
request: Request,
|
||||
q: str = "",
|
||||
exclude_id: Optional[int] = None,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Search requirements by name or tag code for autocomplete.
|
||||
User must be a member of the project.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
q: Search query (searches tag_code and req_name)
|
||||
exclude_id: Optional requirement ID to exclude from results (to prevent self-linking)
|
||||
|
||||
Returns:
|
||||
List of matching requirements (limited to 20).
|
||||
"""
|
||||
user = await _get_current_user_db(request, db)
|
||||
await _verify_project_membership(project_id, user.id, db)
|
||||
|
||||
req_repo = RequirementRepository(db)
|
||||
requirements = await req_repo.search_by_name_or_tag(
|
||||
project_id=project_id,
|
||||
query=q,
|
||||
exclude_id=exclude_id,
|
||||
limit=20
|
||||
)
|
||||
|
||||
return [
|
||||
RequirementSearchResult(
|
||||
id=req.id,
|
||||
req_name=req.req_name,
|
||||
tag_code=req.tag.tag_code
|
||||
)
|
||||
for req in requirements
|
||||
]
|
||||
|
||||
|
||||
# ===========================================
|
||||
# Requirement Links Endpoints
|
||||
# ===========================================
|
||||
|
||||
@app.get("/api/requirements/{requirement_id}/links", response_model=List[RequirementLinkResponse])
|
||||
async def get_requirement_links(
|
||||
requirement_id: int,
|
||||
request: Request,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Get all links for a requirement (both outgoing and incoming).
|
||||
User must be a member of the requirement's project.
|
||||
|
||||
Args:
|
||||
requirement_id: The requirement ID
|
||||
|
||||
Returns:
|
||||
List of links with direction 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"
|
||||
)
|
||||
|
||||
await _verify_project_membership(requirement.project_id, user.id, db)
|
||||
|
||||
link_repo = RequirementLinkRepository(db)
|
||||
links = await link_repo.get_by_requirement_id(requirement_id)
|
||||
|
||||
return [RequirementLinkResponse(**link) for link in links]
|
||||
|
||||
|
||||
@app.post("/api/requirements/{requirement_id}/links", response_model=RequirementLinkResponse, status_code=status.HTTP_201_CREATED)
|
||||
async def create_requirement_link(
|
||||
requirement_id: int,
|
||||
request: Request,
|
||||
link_data: RequirementLinkCreateRequest,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Create a new link from this requirement to another.
|
||||
User must be a member of the project.
|
||||
Auditors (role_id=2) cannot create links.
|
||||
|
||||
Args:
|
||||
requirement_id: The source requirement ID
|
||||
link_data: The target requirement and relationship type
|
||||
|
||||
Returns:
|
||||
The created link.
|
||||
"""
|
||||
user = await _get_current_user_db(request, db)
|
||||
|
||||
# Auditors cannot create links
|
||||
_require_role(user, [1, 3], "create requirement links")
|
||||
|
||||
req_repo = RequirementRepository(db)
|
||||
|
||||
# Check if source requirement exists
|
||||
source_req = await req_repo.get_by_id(requirement_id)
|
||||
if not source_req:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Requirement with id {requirement_id} not found"
|
||||
)
|
||||
|
||||
await _verify_project_membership(source_req.project_id, user.id, db)
|
||||
|
||||
# Check if target requirement exists
|
||||
target_req = await req_repo.get_by_id(link_data.target_requirement_id)
|
||||
if not target_req:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Target requirement with id {link_data.target_requirement_id} not found"
|
||||
)
|
||||
|
||||
# Prevent self-linking
|
||||
if requirement_id == link_data.target_requirement_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Cannot link a requirement to itself"
|
||||
)
|
||||
|
||||
# Verify both requirements are in the same project
|
||||
if source_req.project_id != target_req.project_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Cannot link requirements from different projects"
|
||||
)
|
||||
|
||||
# Verify relationship type exists and belongs to the project
|
||||
rel_type_repo = RelationshipTypeRepository(db)
|
||||
rel_type = await rel_type_repo.get_by_id(link_data.relationship_type_id)
|
||||
if not rel_type:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Relationship type with id {link_data.relationship_type_id} not found"
|
||||
)
|
||||
if rel_type.project_id != source_req.project_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Relationship type does not belong to this project"
|
||||
)
|
||||
|
||||
# Check if link already exists
|
||||
link_repo = RequirementLinkRepository(db)
|
||||
if await link_repo.link_exists(requirement_id, link_data.target_requirement_id, link_data.relationship_type_id):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="This link already exists"
|
||||
)
|
||||
|
||||
# Create the link
|
||||
link = await link_repo.create(
|
||||
source_req_id=requirement_id,
|
||||
target_req_id=link_data.target_requirement_id,
|
||||
relationship_type_id=link_data.relationship_type_id,
|
||||
created_by=user.id
|
||||
)
|
||||
|
||||
await db.commit()
|
||||
|
||||
return RequirementLinkResponse(
|
||||
id=link.id,
|
||||
direction="outgoing",
|
||||
type_name=rel_type.type_name,
|
||||
type_id=link.relationship_type_id,
|
||||
inverse_type_name=rel_type.inverse_type_name,
|
||||
linked_requirement={
|
||||
"id": target_req.id,
|
||||
"req_name": target_req.req_name,
|
||||
"tag_code": target_req.tag.tag_code
|
||||
},
|
||||
created_by_username=user.sub,
|
||||
created_by_id=user.id,
|
||||
created_at=link.created_at
|
||||
)
|
||||
|
||||
|
||||
@app.delete("/api/requirement-links/{link_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_requirement_link(
|
||||
link_id: int,
|
||||
request: Request,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Delete a requirement link.
|
||||
Only the creator or an admin can delete a link.
|
||||
|
||||
Args:
|
||||
link_id: The link ID to delete
|
||||
"""
|
||||
user = await _get_current_user_db(request, db)
|
||||
|
||||
link_repo = RequirementLinkRepository(db)
|
||||
link = await link_repo.get_by_id(link_id)
|
||||
|
||||
if not link:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Link with id {link_id} not found"
|
||||
)
|
||||
|
||||
# Verify user is a member of the project
|
||||
await _verify_project_membership(link.source_requirement.project_id, user.id, db)
|
||||
|
||||
# Only creator or admin can delete
|
||||
if link.created_by != user.id and user.role_id != 1:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only the link creator or an admin can delete this link"
|
||||
)
|
||||
|
||||
await link_repo.delete(link_id)
|
||||
await db.commit()
|
||||
|
||||
@@ -193,3 +193,67 @@ class RequirementUpdateRequest(BaseModel):
|
||||
tag_id: Optional[int] = None
|
||||
priority_id: Optional[int] = None
|
||||
group_ids: Optional[List[int]] = None
|
||||
|
||||
|
||||
# Relationship Type schemas
|
||||
class RelationshipTypeResponse(BaseModel):
|
||||
"""Response schema for a relationship type."""
|
||||
id: int
|
||||
project_id: int
|
||||
type_name: str
|
||||
type_description: Optional[str] = None
|
||||
inverse_type_name: Optional[str] = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class RelationshipTypeCreateRequest(BaseModel):
|
||||
"""Request schema for creating a relationship type."""
|
||||
type_name: str
|
||||
type_description: Optional[str] = None
|
||||
inverse_type_name: Optional[str] = None
|
||||
|
||||
|
||||
# Requirement Link schemas
|
||||
class LinkedRequirementInfo(BaseModel):
|
||||
"""Brief info about a linked requirement."""
|
||||
id: int
|
||||
req_name: str
|
||||
tag_code: str
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class RequirementLinkResponse(BaseModel):
|
||||
"""Response schema for a requirement link with direction."""
|
||||
id: int
|
||||
direction: str # 'outgoing' or 'incoming'
|
||||
type_name: str
|
||||
type_id: int
|
||||
inverse_type_name: Optional[str] = None
|
||||
linked_requirement: LinkedRequirementInfo
|
||||
created_by_username: Optional[str] = None
|
||||
created_by_id: Optional[int] = None
|
||||
created_at: Optional[datetime] = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class RequirementLinkCreateRequest(BaseModel):
|
||||
"""Request schema for creating a requirement link."""
|
||||
relationship_type_id: int
|
||||
target_requirement_id: int
|
||||
|
||||
|
||||
# Requirement Search schemas
|
||||
class RequirementSearchResult(BaseModel):
|
||||
"""Response schema for requirement search results (for autocomplete)."""
|
||||
id: int
|
||||
req_name: str
|
||||
tag_code: str
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
@@ -9,6 +9,8 @@ from src.repositories.priority_repository import PriorityRepository
|
||||
from src.repositories.project_repository import ProjectRepository
|
||||
from src.repositories.validation_status_repository import ValidationStatusRepository
|
||||
from src.repositories.validation_repository import ValidationRepository
|
||||
from src.repositories.relationship_type_repository import RelationshipTypeRepository
|
||||
from src.repositories.requirement_link_repository import RequirementLinkRepository
|
||||
|
||||
__all__ = [
|
||||
"UserRepository",
|
||||
@@ -20,4 +22,6 @@ __all__ = [
|
||||
"ProjectRepository",
|
||||
"ValidationStatusRepository",
|
||||
"ValidationRepository",
|
||||
"RelationshipTypeRepository",
|
||||
"RequirementLinkRepository",
|
||||
]
|
||||
|
||||
97
backend/src/repositories/relationship_type_repository.py
Normal file
97
backend/src/repositories/relationship_type_repository.py
Normal file
@@ -0,0 +1,97 @@
|
||||
"""
|
||||
Repository for RelationshipType database operations.
|
||||
"""
|
||||
from typing import List, Optional
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from src.db_models import RelationshipType
|
||||
|
||||
|
||||
class RelationshipTypeRepository:
|
||||
"""Repository for relationship type CRUD operations."""
|
||||
|
||||
def __init__(self, db: AsyncSession):
|
||||
self.db = db
|
||||
|
||||
async def get_by_project_id(self, project_id: int) -> List[RelationshipType]:
|
||||
"""
|
||||
Get all relationship types for a project.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
|
||||
Returns:
|
||||
List of relationship types for the project.
|
||||
"""
|
||||
result = await self.db.execute(
|
||||
select(RelationshipType)
|
||||
.where(RelationshipType.project_id == project_id)
|
||||
.order_by(RelationshipType.type_name)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_by_id(self, relationship_type_id: int) -> Optional[RelationshipType]:
|
||||
"""
|
||||
Get a relationship type by ID.
|
||||
|
||||
Args:
|
||||
relationship_type_id: The relationship type ID
|
||||
|
||||
Returns:
|
||||
The relationship type or None if not found.
|
||||
"""
|
||||
result = await self.db.execute(
|
||||
select(RelationshipType)
|
||||
.where(RelationshipType.id == relationship_type_id)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def create(
|
||||
self,
|
||||
project_id: int,
|
||||
type_name: str,
|
||||
type_description: Optional[str] = None,
|
||||
inverse_type_name: Optional[str] = None
|
||||
) -> RelationshipType:
|
||||
"""
|
||||
Create a new relationship type.
|
||||
|
||||
Args:
|
||||
project_id: The project this type belongs to
|
||||
type_name: Name of the relationship (e.g., "Depends On")
|
||||
type_description: Optional description
|
||||
inverse_type_name: Optional inverse name (e.g., "Depended By")
|
||||
|
||||
Returns:
|
||||
The created relationship type.
|
||||
"""
|
||||
relationship_type = RelationshipType(
|
||||
project_id=project_id,
|
||||
type_name=type_name,
|
||||
type_description=type_description,
|
||||
inverse_type_name=inverse_type_name
|
||||
)
|
||||
self.db.add(relationship_type)
|
||||
await self.db.flush()
|
||||
await self.db.refresh(relationship_type)
|
||||
return relationship_type
|
||||
|
||||
async def delete(self, relationship_type_id: int) -> bool:
|
||||
"""
|
||||
Delete a relationship type by ID.
|
||||
|
||||
Args:
|
||||
relationship_type_id: The ID of the relationship type to delete
|
||||
|
||||
Returns:
|
||||
True if deleted, False if not found.
|
||||
"""
|
||||
result = await self.db.execute(
|
||||
select(RelationshipType).where(RelationshipType.id == relationship_type_id)
|
||||
)
|
||||
relationship_type = result.scalar_one_or_none()
|
||||
if relationship_type:
|
||||
await self.db.delete(relationship_type)
|
||||
await self.db.flush()
|
||||
return True
|
||||
return False
|
||||
192
backend/src/repositories/requirement_link_repository.py
Normal file
192
backend/src/repositories/requirement_link_repository.py
Normal file
@@ -0,0 +1,192 @@
|
||||
"""
|
||||
Repository for RequirementLink database operations.
|
||||
"""
|
||||
from typing import List, Optional, Dict, Any
|
||||
from sqlalchemy import select, or_
|
||||
from sqlalchemy.orm import selectinload
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from src.db_models import RequirementLink, Requirement, RelationshipType, User
|
||||
|
||||
|
||||
class RequirementLinkRepository:
|
||||
"""Repository for requirement link CRUD operations."""
|
||||
|
||||
def __init__(self, db: AsyncSession):
|
||||
self.db = db
|
||||
|
||||
async def get_by_requirement_id(self, requirement_id: int) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all links for a requirement (both outgoing and incoming).
|
||||
Returns a list of dicts with direction info.
|
||||
|
||||
Args:
|
||||
requirement_id: The requirement ID
|
||||
|
||||
Returns:
|
||||
List of link data with direction ('outgoing' or 'incoming').
|
||||
"""
|
||||
# Get outgoing links (this requirement is the source)
|
||||
outgoing_result = await self.db.execute(
|
||||
select(RequirementLink)
|
||||
.options(
|
||||
selectinload(RequirementLink.target_requirement).selectinload(Requirement.tag),
|
||||
selectinload(RequirementLink.relationship_type),
|
||||
selectinload(RequirementLink.creator)
|
||||
)
|
||||
.where(RequirementLink.source_req_id == requirement_id)
|
||||
.order_by(RequirementLink.created_at.desc())
|
||||
)
|
||||
outgoing_links = outgoing_result.scalars().all()
|
||||
|
||||
# Get incoming links (this requirement is the target)
|
||||
incoming_result = await self.db.execute(
|
||||
select(RequirementLink)
|
||||
.options(
|
||||
selectinload(RequirementLink.source_requirement).selectinload(Requirement.tag),
|
||||
selectinload(RequirementLink.relationship_type),
|
||||
selectinload(RequirementLink.creator)
|
||||
)
|
||||
.where(RequirementLink.target_req_id == requirement_id)
|
||||
.order_by(RequirementLink.created_at.desc())
|
||||
)
|
||||
incoming_links = incoming_result.scalars().all()
|
||||
|
||||
# Build result list with direction info
|
||||
result = []
|
||||
|
||||
for link in outgoing_links:
|
||||
result.append({
|
||||
"id": link.id,
|
||||
"direction": "outgoing",
|
||||
"type_name": link.relationship_type.type_name,
|
||||
"type_id": link.relationship_type_id,
|
||||
"inverse_type_name": link.relationship_type.inverse_type_name,
|
||||
"linked_requirement": {
|
||||
"id": link.target_requirement.id,
|
||||
"req_name": link.target_requirement.req_name,
|
||||
"tag_code": link.target_requirement.tag.tag_code
|
||||
},
|
||||
"created_by_username": link.creator.sub if link.creator else None,
|
||||
"created_by_id": link.created_by,
|
||||
"created_at": link.created_at
|
||||
})
|
||||
|
||||
for link in incoming_links:
|
||||
result.append({
|
||||
"id": link.id,
|
||||
"direction": "incoming",
|
||||
"type_name": link.relationship_type.inverse_type_name or link.relationship_type.type_name,
|
||||
"type_id": link.relationship_type_id,
|
||||
"inverse_type_name": link.relationship_type.type_name,
|
||||
"linked_requirement": {
|
||||
"id": link.source_requirement.id,
|
||||
"req_name": link.source_requirement.req_name,
|
||||
"tag_code": link.source_requirement.tag.tag_code
|
||||
},
|
||||
"created_by_username": link.creator.sub if link.creator else None,
|
||||
"created_by_id": link.created_by,
|
||||
"created_at": link.created_at
|
||||
})
|
||||
|
||||
# Sort by created_at descending
|
||||
result.sort(key=lambda x: x["created_at"] or "", reverse=True)
|
||||
return result
|
||||
|
||||
async def get_by_id(self, link_id: int) -> Optional[RequirementLink]:
|
||||
"""
|
||||
Get a link by ID with related data.
|
||||
|
||||
Args:
|
||||
link_id: The link ID
|
||||
|
||||
Returns:
|
||||
The link or None if not found.
|
||||
"""
|
||||
result = await self.db.execute(
|
||||
select(RequirementLink)
|
||||
.options(
|
||||
selectinload(RequirementLink.source_requirement),
|
||||
selectinload(RequirementLink.target_requirement),
|
||||
selectinload(RequirementLink.relationship_type),
|
||||
selectinload(RequirementLink.creator)
|
||||
)
|
||||
.where(RequirementLink.id == link_id)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def create(
|
||||
self,
|
||||
source_req_id: int,
|
||||
target_req_id: int,
|
||||
relationship_type_id: int,
|
||||
created_by: Optional[int] = None
|
||||
) -> RequirementLink:
|
||||
"""
|
||||
Create a new requirement link.
|
||||
|
||||
Args:
|
||||
source_req_id: The source requirement ID
|
||||
target_req_id: The target requirement ID
|
||||
relationship_type_id: The type of relationship
|
||||
created_by: The user creating the link
|
||||
|
||||
Returns:
|
||||
The created link.
|
||||
"""
|
||||
link = RequirementLink(
|
||||
source_req_id=source_req_id,
|
||||
target_req_id=target_req_id,
|
||||
relationship_type_id=relationship_type_id,
|
||||
created_by=created_by
|
||||
)
|
||||
self.db.add(link)
|
||||
await self.db.flush()
|
||||
await self.db.refresh(link)
|
||||
return link
|
||||
|
||||
async def delete(self, link_id: int) -> bool:
|
||||
"""
|
||||
Delete a link by ID.
|
||||
|
||||
Args:
|
||||
link_id: The ID of the link to delete
|
||||
|
||||
Returns:
|
||||
True if deleted, False if not found.
|
||||
"""
|
||||
result = await self.db.execute(
|
||||
select(RequirementLink).where(RequirementLink.id == link_id)
|
||||
)
|
||||
link = result.scalar_one_or_none()
|
||||
if link:
|
||||
await self.db.delete(link)
|
||||
await self.db.flush()
|
||||
return True
|
||||
return False
|
||||
|
||||
async def link_exists(
|
||||
self,
|
||||
source_req_id: int,
|
||||
target_req_id: int,
|
||||
relationship_type_id: int
|
||||
) -> bool:
|
||||
"""
|
||||
Check if a link already exists between two requirements with the same type.
|
||||
|
||||
Args:
|
||||
source_req_id: The source requirement ID
|
||||
target_req_id: The target requirement ID
|
||||
relationship_type_id: The relationship type ID
|
||||
|
||||
Returns:
|
||||
True if link exists, False otherwise.
|
||||
"""
|
||||
result = await self.db.execute(
|
||||
select(RequirementLink)
|
||||
.where(
|
||||
RequirementLink.source_req_id == source_req_id,
|
||||
RequirementLink.target_req_id == target_req_id,
|
||||
RequirementLink.relationship_type_id == relationship_type_id
|
||||
)
|
||||
)
|
||||
return result.scalar_one_or_none() is not None
|
||||
@@ -300,3 +300,52 @@ class RequirementRepository:
|
||||
await self.session.delete(requirement)
|
||||
await self.session.flush()
|
||||
return True
|
||||
|
||||
async def search_by_name_or_tag(
|
||||
self,
|
||||
project_id: int,
|
||||
query: str,
|
||||
exclude_id: Optional[int] = None,
|
||||
limit: int = 20
|
||||
) -> List[Requirement]:
|
||||
"""
|
||||
Search requirements by name or tag code within a project.
|
||||
Used for autocomplete in the add relationship modal.
|
||||
|
||||
Args:
|
||||
project_id: The project ID to search within
|
||||
query: The search query (searches req_name and tag_code)
|
||||
exclude_id: Optional requirement ID to exclude from results
|
||||
limit: Maximum number of results to return
|
||||
|
||||
Returns:
|
||||
List of matching requirements
|
||||
"""
|
||||
from sqlalchemy import or_, func
|
||||
|
||||
stmt = (
|
||||
select(Requirement)
|
||||
.options(selectinload(Requirement.tag))
|
||||
.join(Requirement.tag)
|
||||
.where(Requirement.project_id == project_id)
|
||||
)
|
||||
|
||||
# Add search filter if query is provided
|
||||
if query:
|
||||
# Trim whitespace and search
|
||||
search_term = f"%{query.strip().lower()}%"
|
||||
stmt = stmt.where(
|
||||
or_(
|
||||
func.lower(Requirement.req_name).like(search_term),
|
||||
func.lower(Tag.tag_code).like(search_term)
|
||||
)
|
||||
)
|
||||
|
||||
# Exclude specific requirement (to prevent self-linking)
|
||||
if exclude_id is not None:
|
||||
stmt = stmt.where(Requirement.id != exclude_id)
|
||||
|
||||
stmt = stmt.order_by(Tag.tag_code, Requirement.req_name).limit(limit)
|
||||
|
||||
result = await self.session.execute(stmt)
|
||||
return list(result.scalars().all())
|
||||
|
||||
Reference in New Issue
Block a user