90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
import datetime
|
|
import uuid
|
|
|
|
from decimal import Decimal
|
|
from typing import TYPE_CHECKING, Optional
|
|
|
|
from sqlalchemy import (
|
|
Boolean,
|
|
Column,
|
|
Date,
|
|
DateTime,
|
|
ForeignKey,
|
|
Index,
|
|
Integer,
|
|
Numeric,
|
|
select,
|
|
)
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from sqlalchemy.orm import Mapped, Session, relationship
|
|
|
|
from .meta import Base
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
# if the target of the relationship is in another module
|
|
# that cannot normally be imported at runtime
|
|
from .employee import Employee
|
|
from .user import User
|
|
|
|
|
|
class Attendance(Base):
|
|
__tablename__ = "attendances"
|
|
id: uuid.UUID = Column("id", UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
employee_id: uuid.UUID = Column("employee_id", UUID(as_uuid=True), ForeignKey("employees.id"))
|
|
date: datetime.date = Column("date", Date, nullable=False)
|
|
attendance_type: int = Column("attendance_type", Integer)
|
|
amount: Decimal = Column("amount", Numeric(precision=5, scale=2))
|
|
creation_date: datetime.datetime = Column("creation_date", DateTime())
|
|
user_id: uuid.UUID = Column("user_id", UUID(as_uuid=True), ForeignKey("users.id"))
|
|
is_valid: bool = Column("is_valid", Boolean)
|
|
|
|
__table_args__ = (
|
|
Index(
|
|
"only_one_valid_attendance",
|
|
"employee_id",
|
|
"date",
|
|
unique=True,
|
|
postgresql_where=(is_valid == True), # noqa: E712
|
|
),
|
|
)
|
|
user: Mapped["User"] = relationship("User")
|
|
employee: Mapped["Employee"] = relationship("Employee", back_populates="attendances")
|
|
|
|
def __init__(
|
|
self,
|
|
employee_id: uuid.UUID,
|
|
date: datetime.date,
|
|
attendance_type: int,
|
|
user_id: uuid.UUID,
|
|
amount: Decimal = Decimal(0),
|
|
creation_date: Optional[datetime.datetime] = None,
|
|
is_valid: bool = True,
|
|
id_: Optional[uuid.UUID] = None,
|
|
):
|
|
self.employee_id = employee_id
|
|
self.date = date
|
|
self.attendance_type = attendance_type
|
|
self.user_id = user_id
|
|
self.amount = amount
|
|
self.creation_date = creation_date or datetime.datetime.utcnow()
|
|
self.is_valid = is_valid
|
|
if id_ is not None:
|
|
self.id = id_
|
|
|
|
def create(self, db: Session) -> None:
|
|
old = (
|
|
db.execute(
|
|
select(Attendance)
|
|
.filter(Attendance.date == self.date)
|
|
.filter(Attendance.employee_id == self.employee_id)
|
|
.filter(Attendance.is_valid == True) # noqa: E712
|
|
)
|
|
.scalars()
|
|
.one_or_none()
|
|
)
|
|
if old is None or old.attendance_type != self.attendance_type:
|
|
if old is not None:
|
|
old.is_valid = False
|
|
db.add(self)
|