NEtraAi / backend /app /models /models.py
093xpku
Clean project layout deployment
9bc686b
Raw
History Blame
8.76 kB
import datetime
import json
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Date, ForeignKey, Float, Text, Time, Interval, Enum
from sqlalchemy.types import TypeDecorator
from sqlalchemy.orm import relationship
from app.core.database import Base
class SafeVector(TypeDecorator):
impl = Text
cache_ok = True
def __init__(self, dimensions):
self.dimensions = dimensions
super().__init__()
def load_dialect_impl(self, dialect):
if dialect.name == 'postgresql':
try:
from pgvector.sqlalchemy import Vector
return dialect.type_descriptor(Vector(self.dimensions))
except ImportError:
pass
return dialect.type_descriptor(Text())
def process_bind_param(self, value, dialect):
if value is None:
return None
if dialect.name == 'postgresql':
return value
return json.dumps(value)
def process_result_value(self, value, dialect):
if value is None:
return None
if dialect.name == 'postgresql':
return value
try:
return json.loads(value)
except Exception:
return value
class Role(Base):
__tablename__ = "roles"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), unique=True, nullable=False) # Super Admin, Admin, HR, Employee
description = Column(String(255), nullable=True)
users = relationship("User", back_populates="role")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String(255), unique=True, index=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
role_id = Column(Integer, ForeignKey("roles.id"), nullable=False)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
role = relationship("Role", back_populates="users")
employee = relationship("Employee", back_populates="user", uselist=False)
audit_logs = relationship("AuditLog", back_populates="user")
class Department(Base):
__tablename__ = "departments"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), unique=True, nullable=False)
code = Column(String(20), unique=True, nullable=False)
description = Column(String(255), nullable=True)
employees = relationship("Employee", back_populates="department")
class Employee(Base):
__tablename__ = "employees"
id = Column(Integer, primary_key=True, index=True)
employee_id = Column(String(50), unique=True, index=True, nullable=False)
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, index=True, nullable=False)
phone = Column(String(20), nullable=True)
designation = Column(String(100), nullable=True)
joining_date = Column(Date, nullable=False, default=datetime.date.today)
status = Column(String(20), default="Active") # Active, Inactive, Suspended
department_id = Column(Integer, ForeignKey("departments.id"), nullable=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, unique=True)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
department = relationship("Department", back_populates="employees")
user = relationship("User", back_populates="employee")
images = relationship("EmployeeImage", back_populates="employee", cascade="all, delete-orphan")
embeddings = relationship("FaceEmbedding", back_populates="employee", cascade="all, delete-orphan")
attendance_records = relationship("Attendance", back_populates="employee", cascade="all, delete-orphan")
attendance_logs = relationship("AttendanceLog", back_populates="employee", cascade="all, delete-orphan")
leave_requests = relationship("LeaveRequest", back_populates="employee", cascade="all, delete-orphan")
class EmployeeImage(Base):
__tablename__ = "employee_images"
id = Column(Integer, primary_key=True, index=True)
employee_id = Column(Integer, ForeignKey("employees.id", ondelete="CASCADE"), nullable=False)
file_path = Column(String(255), nullable=False)
pose_type = Column(String(50), nullable=False) # Front, Left, Right, Up, Down, Smile, Neutral, etc.
created_at = Column(DateTime, default=datetime.datetime.utcnow)
employee = relationship("Employee", back_populates="images")
embeddings = relationship("FaceEmbedding", back_populates="image", cascade="all, delete-orphan")
class FaceEmbedding(Base):
__tablename__ = "face_embeddings"
id = Column(Integer, primary_key=True, index=True)
employee_id = Column(Integer, ForeignKey("employees.id", ondelete="CASCADE"), nullable=False)
image_id = Column(Integer, ForeignKey("employee_images.id", ondelete="CASCADE"), nullable=True)
embedding = Column(SafeVector(512), nullable=False) # pgvector field for 512 dimensions (ArcFace) or JSON text on SQLite
created_at = Column(DateTime, default=datetime.datetime.utcnow)
employee = relationship("Employee", back_populates="embeddings")
image = relationship("EmployeeImage", back_populates="embeddings")
class Attendance(Base):
__tablename__ = "attendance"
id = Column(Integer, primary_key=True, index=True)
employee_id = Column(Integer, ForeignKey("employees.id", ondelete="CASCADE"), nullable=False)
date = Column(Date, nullable=False)
check_in = Column(DateTime, nullable=True)
check_out = Column(DateTime, nullable=True)
working_hours = Column(Float, default=0.0) # Calculated in hours
late_arrival = Column(Boolean, default=False)
early_departure = Column(Boolean, default=False)
overtime = Column(Float, default=0.0) # Calculated in hours
status = Column(String(20), default="Absent") # Present, Absent, Late, Half Day, Leave, Holiday, WFH
employee = relationship("Employee", back_populates="attendance_records")
class AttendanceLog(Base):
__tablename__ = "attendance_logs"
id = Column(Integer, primary_key=True, index=True)
employee_id = Column(Integer, ForeignKey("employees.id", ondelete="CASCADE"), nullable=True) # Null if not recognized
timestamp = Column(DateTime, default=datetime.datetime.now, nullable=False)
camera = Column(String(100), default="Kiosk")
confidence = Column(Float, nullable=True)
liveness_score = Column(Float, nullable=True)
is_spoof = Column(Boolean, default=False)
status = Column(String(50), nullable=False) # Match Success, Spoof Rejected, Unknown Person, Low Confidence
employee = relationship("Employee", back_populates="attendance_logs")
class LeaveRequest(Base):
__tablename__ = "leave_requests"
id = Column(Integer, primary_key=True, index=True)
employee_id = Column(Integer, ForeignKey("employees.id", ondelete="CASCADE"), nullable=False)
start_date = Column(Date, nullable=False)
end_date = Column(Date, nullable=False)
leave_type = Column(String(50), nullable=False) # Sick, Casual, Annual, unpaid
reason = Column(Text, nullable=True)
status = Column(String(20), default="Pending") # Pending, Approved, Rejected
approved_by = Column(Integer, ForeignKey("users.id"), nullable=True)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
employee = relationship("Employee", back_populates="leave_requests")
class Holiday(Base):
__tablename__ = "holidays"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False)
date = Column(Date, unique=True, nullable=False)
description = Column(String(255), nullable=True)
class Setting(Base):
__tablename__ = "settings"
id = Column(Integer, primary_key=True, index=True)
key = Column(String(100), unique=True, nullable=False)
value = Column(Text, nullable=False)
description = Column(String(255), nullable=True)
class AuditLog(Base):
__tablename__ = "audit_logs"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
action = Column(String(100), nullable=False) # Login, Logout, Create Employee, Mark Attendance, etc.
timestamp = Column(DateTime, default=datetime.datetime.now)
ip_address = Column(String(50), nullable=True)
user_agent = Column(String(255), nullable=True)
details = Column(Text, nullable=True)
user = relationship("User", back_populates="audit_logs")