225 lines
7.5 KiB
Python
225 lines
7.5 KiB
Python
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
|
|
from app import db
|
|
|
|
|
|
class Card(db.Model):
|
|
"""Card model for Kanban cards"""
|
|
|
|
__tablename__ = "cards"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(200), nullable=False)
|
|
description = db.Column(db.Text)
|
|
closed = db.Column(db.Boolean, default=False)
|
|
due = db.Column(db.DateTime)
|
|
due_complete = db.Column(db.Boolean, default=False)
|
|
pos = db.Column(db.Float) # position for sorting
|
|
id_short = db.Column(db.Integer) # short ID for URLs
|
|
|
|
# Foreign keys
|
|
board_id = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey("boards.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
list_id = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey("lists.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
epic_id = db.Column(
|
|
db.Integer, db.ForeignKey("epics.id", ondelete="SET NULL"), index=True
|
|
)
|
|
|
|
# Timestamps
|
|
date_last_activity = db.Column(db.DateTime)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
|
|
updated_at = db.Column(
|
|
db.DateTime,
|
|
default=lambda: datetime.now(UTC),
|
|
onupdate=lambda: datetime.now(UTC),
|
|
)
|
|
|
|
# JSON fields
|
|
badges = db.Column(JSONB) # card badges/stats
|
|
cover = db.Column(JSONB) # cover settings
|
|
desc_data = db.Column(JSONB)
|
|
|
|
# Relationships
|
|
checklists = db.relationship(
|
|
"Checklist", backref="card", cascade="all, delete-orphan", lazy="dynamic"
|
|
)
|
|
labels = db.relationship(
|
|
"CardLabel", backref="card", cascade="all, delete-orphan", lazy="dynamic"
|
|
)
|
|
comments = db.relationship(
|
|
"Comment", backref="card", cascade="all, delete-orphan", lazy="dynamic"
|
|
)
|
|
attachments = db.relationship(
|
|
"FileAttachment",
|
|
foreign_keys="FileAttachment.attachable_id",
|
|
primaryjoin="""and_(FileAttachment.attachable_id == Card.id,
|
|
FileAttachment.attachable_type == 'Card')""",
|
|
cascade="all, delete-orphan",
|
|
lazy="dynamic",
|
|
)
|
|
|
|
def to_dict(self):
|
|
"""Convert card to dictionary"""
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"closed": self.closed,
|
|
"due": self.due.isoformat() if self.due else None,
|
|
"due_complete": self.due_complete,
|
|
"pos": self.pos,
|
|
"id_short": self.id_short,
|
|
"board_id": self.board_id,
|
|
"list_id": self.list_id,
|
|
"list_name": self.list.name,
|
|
"epic_id": self.epic_id,
|
|
"date_last_activity": self.date_last_activity.isoformat()
|
|
if self.date_last_activity
|
|
else None,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
"badges": self.badges,
|
|
"cover": self.cover,
|
|
"desc_data": self.desc_data,
|
|
}
|
|
|
|
def __repr__(self):
|
|
return f"<Card {self.name}>"
|
|
|
|
|
|
# SQLAlchemy event listeners to update Epic metrics
|
|
def update_epic_metrics_on_card_change(mapper, connection, target):
|
|
"""Update epic card_count when card epic_id changes"""
|
|
if target.epic_id:
|
|
from sqlalchemy import select, update
|
|
|
|
from app.models import Epic
|
|
|
|
# Get total card count
|
|
card_count_stmt = select(db.func.count(Card.id)).where(
|
|
Card.epic_id == target.epic_id
|
|
)
|
|
card_count = connection.execute(card_count_stmt).scalar()
|
|
|
|
# Get epic's completed_list_id
|
|
completed_list_id_stmt = select(Epic.completed_list_id).where(
|
|
Epic.id == target.epic_id
|
|
)
|
|
completed_list_id = connection.execute(completed_list_id_stmt).scalar()
|
|
|
|
# Get completed card count (only if epic has completed_list_id)
|
|
completed_cards_count = 0
|
|
if completed_list_id:
|
|
completed_cards_stmt = select(db.func.count(Card.id)).where(
|
|
Card.epic_id == target.epic_id, Card.list_id == completed_list_id
|
|
)
|
|
completed_cards_count = connection.execute(completed_cards_stmt).scalar()
|
|
|
|
# Update epic metrics
|
|
connection.execute(
|
|
update(Epic)
|
|
.where(Epic.id == target.epic_id)
|
|
.values(
|
|
metrics={
|
|
"card_count": card_count,
|
|
"completed_cards_count": completed_cards_count,
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
def update_epic_metrics_on_card_insert(mapper, connection, target):
|
|
"""Update epic card_count when a card is added to an epic"""
|
|
if target.epic_id:
|
|
from sqlalchemy import select, update
|
|
|
|
from app.models import Epic
|
|
|
|
# Get total card count
|
|
card_count_stmt = select(db.func.count(Card.id)).where(
|
|
Card.epic_id == target.epic_id
|
|
)
|
|
card_count = connection.execute(card_count_stmt).scalar()
|
|
|
|
# Get epic's completed_list_id
|
|
completed_list_id_stmt = select(Epic.completed_list_id).where(
|
|
Epic.id == target.epic_id
|
|
)
|
|
completed_list_id = connection.execute(completed_list_id_stmt).scalar()
|
|
|
|
# Get completed card count (only if epic has completed_list_id)
|
|
completed_cards_count = 0
|
|
if completed_list_id:
|
|
completed_cards_stmt = select(db.func.count(Card.id)).where(
|
|
Card.epic_id == target.epic_id, Card.list_id == completed_list_id
|
|
)
|
|
completed_cards_count = connection.execute(completed_cards_stmt).scalar()
|
|
|
|
# Update epic metrics
|
|
connection.execute(
|
|
update(Epic)
|
|
.where(Epic.id == target.epic_id)
|
|
.values(
|
|
metrics={
|
|
"card_count": card_count,
|
|
"completed_cards_count": completed_cards_count,
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
def update_epic_metrics_on_card_delete(mapper, connection, target):
|
|
"""Update epic card_count when a card is removed from an epic"""
|
|
if target.epic_id:
|
|
from sqlalchemy import select, update
|
|
|
|
from app.models import Epic
|
|
|
|
# Get total card count
|
|
card_count_stmt = select(db.func.count(Card.id)).where(
|
|
Card.epic_id == target.epic_id
|
|
)
|
|
card_count = connection.execute(card_count_stmt).scalar()
|
|
|
|
# Get epic's completed_list_id
|
|
completed_list_id_stmt = select(Epic.completed_list_id).where(
|
|
Epic.id == target.epic_id
|
|
)
|
|
completed_list_id = connection.execute(completed_list_id_stmt).scalar()
|
|
|
|
# Get completed card count (only if epic has completed_list_id)
|
|
completed_cards_count = 0
|
|
if completed_list_id:
|
|
completed_cards_stmt = select(db.func.count(Card.id)).where(
|
|
Card.epic_id == target.epic_id, Card.list_id == completed_list_id
|
|
)
|
|
completed_cards_count = connection.execute(completed_cards_stmt).scalar()
|
|
|
|
# Update epic metrics
|
|
connection.execute(
|
|
update(Epic)
|
|
.where(Epic.id == target.epic_id)
|
|
.values(
|
|
metrics={
|
|
"card_count": card_count,
|
|
"completed_cards_count": completed_cards_count,
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
# Register event listeners
|
|
db.event.listen(Card, "after_update", update_epic_metrics_on_card_change)
|
|
db.event.listen(Card, "after_insert", update_epic_metrics_on_card_insert)
|
|
db.event.listen(Card, "after_delete", update_epic_metrics_on_card_delete)
|