Skip to content

B
BookingMyHrManagement
Commit cf322a8e by Ooh-Ao
Options

delete

parent 60521a09
Showing with 0 additions and 1359 deletions
API-Fast/src/controllers/company_controller.py deleted 100644 → 0
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from ..models.company import Company
from fastapi import HTTPException, status
import bcrypt
# ฟังก์ชันเพื่อดึงข้อมูลบริษัททั้งหมด
async def get_all_companies(db: AsyncSession):
result = await db.execute(select(Company))
return result.scalars().all()
# ฟังก์ชันเพื่อดึงข้อมูลบริษัทตาม ID
async def get_company_by_id(companyId: str, db: AsyncSession):
result = await db.execute(select(Company).where(Company.companyId == companyId))
company = result.scalar_one_or_none()
if not company:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Company not found")
return company
# ฟังก์ชันเพื่อสร้างบริษัทใหม่
async def create_company(company_data: dict, db: AsyncSession):
new_company = Company(**company_data)
db.add(new_company)
await db.commit()
await db.refresh(new_company)
return new_company
# ฟังก์ชันเพื่ออัปเดตข้อมูลบริษัท
async def update_company(companyId: str, company_data: dict, db: AsyncSession):
company = await get_company_by_id(companyId, db)
for key, value in company_data.items():
setattr(company, key, value)
await db.commit()
await db.refresh(company)
return company
# ฟังก์ชันเพื่อการลบบริษัท
async def delete_company(companyId: str, db: AsyncSession):
company = await get_company_by_id(companyId, db)
await db.delete(company)
await db.commit()
return {
"actionStatus": "Success",
"message": "User deleted successfully",
"statusCode": 200,
}
API-Fast/src/controllers/company_employee_controller.py deleted 100644 → 0
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload
from ..schemas.company_employee_schema import ProjectEmployeePost, ProjectEmployeeUpdate
from ..models.company_employee import ProjectEmployee
from ..models.member import Member, RoleType
from ..models.company import Company
from ..models.position import Position
from ..models.department import Department
from fastapi import Depends, HTTPException, status
from ..config.database import get_db
from ..schemas.member_schema import MemberCreate, MemberResponse
from uuid import UUID, uuid4
# ดึงข้อมูล ProjectEmployee ทั้งหมดตาม companyId
async def get_all_ProjectEmployees(companyId: str, db: AsyncSession):
result = await db.execute(
select(ProjectEmployee)
.where(ProjectEmployee.companyId == companyId)
.options(
selectinload(ProjectEmployee.member),
selectinload(ProjectEmployee.position),
selectinload(ProjectEmployee.department)
)
)
employees = result.scalars().all()
# จัดรูปแบบ response
response = []
for employee in employees:
response.append({
"member": {
"memberId": employee.member.memberId,
"username": employee.member.username,
"email": employee.member.email,
"firstName": employee.member.firstName,
"lastName": employee.member.lastName,
"picture": employee.member.picture
},
"companyId": employee.companyId,
"employeeId": employee.employeeId,
"position": {
"positionId": employee.position.positionId,
"thName": employee.position.thName,
"engName": employee.position.engName
} if employee.position else None,
"department": {
"departmentId": employee.department.departmentId,
"thName": employee.department.thName,
"engName": employee.department.engName
} if employee.department else None,
"company_employeeId": employee.company_employeeId,
"salary" : employee.salary,
"startDate" : employee.startDate,
"companyRoleType" : employee.companyRoleType,
"empType" : employee.empType,
"bossId" : employee.bossId
})
return response
# ดึงข้อมูล ProjectEmployee ตาม memberid และ companyId
async def get_company_member_by_id(memberId: str, companyId: str, db: AsyncSession):
result = await db.execute(
select(ProjectEmployee)
.where(
ProjectEmployee.memberId == memberId,
ProjectEmployee.companyId == companyId
)
.options(
selectinload(ProjectEmployee.member),
selectinload(ProjectEmployee.company),
selectinload(ProjectEmployee.position),
selectinload(ProjectEmployee.department)
)
)
company_employee = result.scalar_one_or_none()
if not company_employee:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Company Employee not found")
return company_employee
# ดึงข้อมูล ProjectEmployee ตาม company_employeeId และ companyId
async def get_company_employee_by_id(company_employeeId: str, companyId: str, db: AsyncSession):
result = await db.execute(
select(ProjectEmployee)
.where(
ProjectEmployee.company_employeeId == company_employeeId,
ProjectEmployee.companyId == companyId
)
.options(
selectinload(ProjectEmployee.member),
selectinload(ProjectEmployee.company),
selectinload(ProjectEmployee.position),
selectinload(ProjectEmployee.department)
)
)
company_employee = result.scalar_one_or_none()
if not company_employee:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Company Employee not found")
return company_employee
# ดึงข้อมูล ProjectEmployee ตาม company_employeeId และ companyId
async def get_company_sub_employee_by_id(company_employeeId: str, companyId: str, db: AsyncSession):
result = await db.execute(
select(ProjectEmployee)
.where(ProjectEmployee.companyId == companyId , ProjectEmployee.bossId == company_employeeId)
.options(
selectinload(ProjectEmployee.member),
selectinload(ProjectEmployee.position),
selectinload(ProjectEmployee.department)
)
)
employees = result.scalars().all()
# จัดรูปแบบ response
response = []
for employee in employees:
response.append({
"member": {
"memberId": employee.member.memberId,
"username": employee.member.username,
"email": employee.member.email,
"firstName": employee.member.firstName,
"lastName": employee.member.lastName,
"picture": employee.member.picture
},
"companyId": employee.companyId,
"employeeId": employee.employeeId,
"position": {
"positionId": employee.position.positionId,
"thName": employee.position.thName,
"engName": employee.position.engName
} if employee.position else None,
"department": {
"departmentId": employee.department.departmentId,
"thName": employee.department.thName,
"engName": employee.department.engName
} if employee.department else None,
"company_employeeId": employee.company_employeeId,
"salary" : employee.salary,
"startDate" : employee.startDate,
"companyRoleType" : employee.companyRoleType,
"empType" : employee.empType,
"bossId" : employee.bossId
})
return response
async def register_member(
emp_data: ProjectEmployeePost, companyId: str ,db: AsyncSession
):
# ตรวจสอบว่าบริษัทมีอยู่จริงหรือไม่
company = await db.get(Company, companyId)
if not company:
raise HTTPException(
status_code=404,
detail="Company not found"
)
# ตรวจสอบว่ามี Member ในระบบหรือไม่
result = await db.execute(
select(Member).where(Member.email == emp_data.member.email)
)
member = result.scalar_one_or_none()
if not member:
# ถ้าไม่มี Member สร้างใหม่
member = Member(
username=emp_data.member.email,
email=emp_data.member.email,
firstName=emp_data.member.firstName,
lastName=emp_data.member.lastName,
picture=emp_data.member.picture,
role=RoleType.MEMBER,
status='ENABLE'
)
member.hash_password("1234")
db.add(member)
await db.commit()
await db.refresh(member)
# ตรวจสอบว่าพนักงานมีอยู่ในบริษัทนี้แล้วหรือไม่
result = await db.execute(
select(ProjectEmployee).where(
ProjectEmployee.memberId == member.memberId,
ProjectEmployee.companyId == company.companyId
)
)
existing_employee = result.scalar_one_or_none()
if existing_employee:
raise HTTPException(
status_code=400,
detail="Member is already registered in this company"
)
# เพิ่มพนักงานในบริษัท
new_employee = ProjectEmployee(
memberId=member.memberId,
companyId=company.companyId,
positionId = emp_data.position.positionId,
departmentId = emp_data.department.departmentId,
employeeId = emp_data.employeeId,
bossId = emp_data.bossId,
salary = emp_data.salary,
companyRoleType = emp_data.companyRoleType,
empType = emp_data.empType,
startDate = emp_data.startDate
)
db.add(new_employee)
await db.commit()
await db.refresh(new_employee)
return {
"member": {
"memberId": str(member.memberId),
"email": member.email,
"firstName": member.firstName,
"lastName": member.lastName,
"picture": member.picture,
},
"position": emp_data.position.dict() if emp_data.position else None,
"department": emp_data.department.dict() if emp_data.department else None,
"companyId" : company.companyId,
"company_employeeId": str(new_employee.company_employeeId),
"employeeId": new_employee.employeeId,
"salary": new_employee.salary,
"bossId": new_employee.bossId,
"companyRoleType": new_employee.companyRoleType,
"empType": new_employee.empType,
"startDate": new_employee.startDate,
}
async def update_company_employee(
company_employee_data: ProjectEmployeeUpdate,
companyId: str,
db: AsyncSession
):
# ดึงข้อมูล ProjectEmployee
company_employee = await get_company_employee_by_id(company_employee_data.company_employeeId, companyId, db)
# รับข้อมูลที่เกี่ยวข้อง
member_info = company_employee_data.member
position_info = company_employee_data.position
department_info = company_employee_data.department
# อัปเดตข้อมูล ProjectEmployee
# for key, value in company_employee_data.items():
# if hasattr(company_employee, key) and value is not None:
# setattr(company_employee, key, value)
if company_employee:
company_employee.employeeId = company_employee_data.employeeId
company_employee.salary = company_employee_data.salary
company_employee.bossId = company_employee_data.bossId
company_employee.companyRoleType = company_employee_data.companyRoleType
company_employee.empType = company_employee_data.empType
company_employee.startDate = company_employee_data.startDate
# อัปเดตข้อมูล Member
if member_info:
member = await db.get(Member, company_employee.memberId)
if member.firstName:
member.firstName = company_employee_data.member.firstName
if member.lastName:
member.lastName = company_employee_data.member.lastName
if member.email:
member.email = company_employee_data.member.email
if member.picture:
member.picture = company_employee_data.member.picture
await db.commit() # บันทึกการเปลี่ยนแปลงของ Member
await db.refresh(member) # โหลดข้อมูล Member ที่อัปเดต
# อัปเดตข้อมูล Position
if position_info:
company_employee.positionId = position_info.positionId
# อัปเดตข้อมูล Department
if department_info:
company_employee.departmentId = department_info.departmentId
await db.commit() # บันทึกการเปลี่ยนแปลงของ ProjectEmployee
await db.refresh(company_employee) # โหลดข้อมูล ProjectEmployee ที่อัปเดต
# จัดรูปแบบ Response
return {
"member": {
"memberId": company_employee.member.memberId,
"username": company_employee.member.username,
"email": company_employee.member.email,
"firstName": company_employee.member.firstName,
"lastName": company_employee.member.lastName,
"picture": company_employee.member.picture,
},
"companyId": company_employee.companyId,
"position": position_info,
"department": department_info,
"company_employeeId": company_employee.company_employeeId,
"employeeId": company_employee.employeeId,
"salary": company_employee.salary,
"bossId": company_employee.bossId,
"companyRoleType": company_employee.companyRoleType,
"empType": company_employee.empType,
"startDate": company_employee.startDate,
}
# ลบ ProjectEmployee ตาม company_employeeId และ companyId
async def delete_company_employee(company_employeeId: str, companyId: str, db: AsyncSession):
company_employee = await get_company_employee_by_id(company_employeeId, companyId, db)
await db.delete(company_employee)
await db.commit()
# ค้นหาบริษัทที่ memberId อยู่
async def get_companies_by_member(memberId: str, db: AsyncSession):
result = await db.execute(
select(Company)
.join(ProjectEmployee, Company.companyId == ProjectEmployee.companyId)
.where(ProjectEmployee.memberId == memberId),
)
companies = result.scalars().all()
if not companies:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="No companies found for the given memberId"
)
return companies
# ค้นหาบริษัทที่ memberId อยู่
async def get_companies_admin(memberId: str, db: AsyncSession):
result = await db.execute(
select(Company)
.join(ProjectEmployee, Company.companyId == ProjectEmployee.companyId)
.where(ProjectEmployee.memberId == memberId,
ProjectEmployee.companyRoleType == 'ADMIN_COMPANY'
),
)
companies = result.scalars().all()
if not companies:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="No companies found for Admin"
)
return companies
\ No newline at end of file
API-Fast/src/controllers/company_location_controller.py deleted 100644 → 0
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from fastapi import HTTPException, status
from uuid import UUID
from ..models.company_location import CompanyLocation
# ดึงข้อมูลสถานที่ทั้งหมดของบริษัทตาม companyId
async def get_all_locations(companyId: UUID, db: AsyncSession):
result = await db.execute(
select(CompanyLocation).where(CompanyLocation.companyId == companyId)
)
return result.scalars().all()
# ดึงข้อมูลสถานที่ตาม locationId และ companyId
async def get_location_by_id(locationId: UUID, companyId: UUID, db: AsyncSession):
result = await db.execute(
select(CompanyLocation).where(
CompanyLocation.locationId == locationId,
CompanyLocation.companyId == companyId
)
)
location = result.scalar_one_or_none()
if not location:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Location not found")
return location
# สร้างสถานที่ใหม่
async def create_location(data: dict, companyId: UUID, db: AsyncSession):
data["companyId"] = companyId
new_location = CompanyLocation(**data)
db.add(new_location)
await db.commit()
await db.refresh(new_location)
return new_location
# อัปเดตข้อมูลสถานที่
async def update_location(locationId: UUID, companyId: UUID, data: dict, db: AsyncSession):
location = await get_location_by_id(locationId, companyId, db)
for key, value in data.items():
setattr(location, key, value)
await db.commit()
await db.refresh(location)
return location
# ลบสถานที่
async def delete_location(locationId: UUID, companyId: UUID, db: AsyncSession):
location = await get_location_by_id(locationId, companyId, db)
await db.delete(location)
await db.commit()
API-Fast/src/controllers/department_controller.py deleted 100644 → 0
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import joinedload
from ..models.department import Department
from ..models.company import Company
from fastapi import HTTPException, status
# ดึงข้อมูล department ทั้งหมด
async def get_all_departments(db: AsyncSession):
result = await db.execute(select(Department).options(joinedload(Department.company)))
return result.scalars().all()
# ดึงข้อมูล department ตาม ID
async def get_department_by_id(departmentId: str, db: AsyncSession):
result = await db.execute(select(Department).where(Department.departmentId == departmentId).options(joinedload(Department.company)))
department = result.scalar_one_or_none()
if not department:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Department not found")
return department
# ดึงข้อมูล departments โดยใช้ companyId
async def get_departments_by_companyId(companyId: str, db: AsyncSession):
result = await db.execute(select(Department).where(Department.companyId == companyId))
return result.scalars().all()
# สร้าง department ใหม่
async def create_department(department_data: dict, db: AsyncSession):
# ตรวจสอบว่ามี companyId ที่อ้างอิงถึงอยู่หรือไม่
companyId = department_data["companyId"]
company = await db.get(Company, companyId)
if not company:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Company not found")
new_department = Department(**department_data)
db.add(new_department)
await db.commit()
await db.refresh(new_department)
return new_department
# อัปเดต department
async def update_department(departmentId: str, department_data: dict, db: AsyncSession):
department = await db.get(Department, departmentId)
if not department:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Department not found")
# อัปเดตข้อมูล department
for key, value in department_data.items():
setattr(department, key, value)
await db.commit()
await db.refresh(department)
return department
# ลบ department
async def delete_department(departmentId: str, db: AsyncSession):
department = await db.get(Department, departmentId)
if not department:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Department not found")
await db.delete(department)
await db.commit()
API-Fast/src/controllers/employee_timestamp_controller.py deleted 100644 → 0
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload
from fastapi import HTTPException, status
from typing import Optional, List
from datetime import datetime
from ..schemas.employee_timestamp_schema import EmployeeTimestampUpdate
from ..models.employee_timestamp import EmployeeTimestamp, TimestampType
from uuid import UUID
from ..models.company_employee import ProjectEmployee
# ดึงข้อมูลการลงเวลาทั้งหมดของพนักงานในบริษัท
async def get_all_timestamps(companyId: str, db: AsyncSession):
result = await db.execute(
select(EmployeeTimestamp).where(EmployeeTimestamp.companyId == companyId).options(
selectinload(EmployeeTimestamp.company_employee).joinedload(ProjectEmployee.member),
selectinload(EmployeeTimestamp.company_employee).joinedload(ProjectEmployee.position),
selectinload(EmployeeTimestamp.company_employee).joinedload(ProjectEmployee.department)
)
)
timestamps = result.scalars().all()
# จัดรูปแบบ response
response = []
for time in timestamps:
response.append({
"employee" : time.company_employee,
"company_employeeId": time.company_employeeId,
"timestampType": time.timestampType,
"locationName" : time.locationName,
"latitude": time.latitude,
"longitude": time.longitude,
"photoTimestamp": time.photoTimestamp,
"timestampId": time.timestampId,
"companyId": time.companyId,
"timestamp": time.timestamp,
"createdAt": time.createdAt,
"updatedAt": time.updatedAt
})
return response
# ดึงข้อมูลการลงเวลาของพนักงานจาก timestampId
async def get_timestamp_by_id(timestampId: UUID, companyId: UUID, db: AsyncSession):
result = await db.execute(
select(EmployeeTimestamp).where(
EmployeeTimestamp.timestampId == timestampId,
EmployeeTimestamp.companyId == companyId
)
)
timestamp = result.scalar_one_or_none()
if not timestamp:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Timestamp not found")
return timestamp
# สร้างข้อมูลการลงเวลาของพนักงานใหม่
async def create_timestamp(data: dict, companyId: UUID, db: AsyncSession):
data["companyId"] = companyId
new_timestamp = EmployeeTimestamp(**data)
db.add(new_timestamp)
await db.commit()
await db.refresh(new_timestamp)
result = await db.execute(
select(ProjectEmployee)
.where(
ProjectEmployee.company_employeeId == new_timestamp.company_employeeId,
ProjectEmployee.companyId == companyId
)
.options(
selectinload(ProjectEmployee.member),
selectinload(ProjectEmployee.company),
selectinload(ProjectEmployee.position),
selectinload(ProjectEmployee.department)
)
)
company_employee = result.scalar_one_or_none()
if not company_employee:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Company Employee not found")
return {
"employee": company_employee,
"company_employeeId": new_timestamp.company_employeeId,
"timestampType": new_timestamp.timestampType,
"locationName" : new_timestamp.locationName,
"latitude": new_timestamp.latitude,
"longitude": new_timestamp.longitude,
"photoTimestamp": new_timestamp.photoTimestamp,
"timestampId": new_timestamp.timestampId,
"companyId": new_timestamp.companyId,
"timestamp": new_timestamp.timestamp,
"createdAt": new_timestamp.createdAt,
"updatedAt": new_timestamp.updatedAt
}
# อัปเดตข้อมูลการลงเวลาของพนักงาน
async def update_timestamp(timestampId: UUID, companyId: UUID, data: EmployeeTimestampUpdate, db: AsyncSession):
timestamp = await get_timestamp_by_id(timestampId, companyId, db)
if data.timestampType:
timestamp.timestampType = data.timestampType
if data.locationName:
timestamp.locationName = data.locationName
await db.commit()
await db.refresh(timestamp)
return timestamp
# ลบข้อมูลการลงเวลาของพนักงาน
async def delete_timestamp(timestampId: UUID, companyId: UUID, db: AsyncSession):
timestamp = await get_timestamp_by_id(timestampId, companyId, db)
await db.delete(timestamp)
await db.commit()
# ค้นหาการลงเวลาตามเงื่อนไข
async def search_timestamps(
companyId: str,
company_employeeId: Optional[str] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
timestampType: Optional[int] = None,
db: AsyncSession = None
):
query = select(EmployeeTimestamp).where(EmployeeTimestamp.companyId == companyId)
if start_date:
query = query.where(EmployeeTimestamp.timestamp >= start_date)
if end_date:
query = query.where(EmployeeTimestamp.timestamp <= end_date)
if company_employeeId:
query = query.where(EmployeeTimestamp.company_employeeId == company_employeeId)
if timestampType:
query = query.where(EmployeeTimestamp.timestampType == timestampType)
result = await db.execute(query.options(
selectinload(EmployeeTimestamp.company_employee).joinedload(ProjectEmployee.member),
selectinload(EmployeeTimestamp.company_employee).joinedload(ProjectEmployee.position),
selectinload(EmployeeTimestamp.company_employee).joinedload(ProjectEmployee.department)
))
timestamps = result.scalars().all()
# จัดรูปแบบ response
response = []
for time in timestamps:
response.append({
"employee": time.company_employee,
"company_employeeId": time.company_employeeId,
"timestampType": time.timestampType,
"locationName" : time.locationName,
"latitude": time.latitude,
"longitude": time.longitude,
"photoTimestamp": time.photoTimestamp,
"timestampId": time.timestampId,
"companyId": time.companyId,
"timestamp": time.timestamp,
"createdAt": time.createdAt,
"updatedAt": time.updatedAt
})
return response
API-Fast/src/controllers/face_controller.py deleted 100644 → 0
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import HTTPException, status
from pathlib import Path
from uuid import UUID
import uuid
import shutil
from ..models.member import Member
from ..utils.face_recognition import add_face_regonite, check_face_regonite,check_face_regonite_emp
from ..utils.file_handler import save_uploaded_file
# ฟังก์ชันสำหรับเพิ่มใบหน้า
async def add_face(memberId: UUID, image_file, db: AsyncSession):
member = await db.get(Member, memberId)
if not member:
raise HTTPException(status_code=404, detail="User not found")
try:
# บันทึกไฟล์ภาพและเก็บเส้นทางของไฟล์
image_path = await save_uploaded_file(image_file)
output = add_face_regonite(str(image_path), str(memberId))
# เก็บข้อมูล faceEmbedding ในฐานข้อมูล
member.faceEmbedding = output
await db.commit()
await db.refresh(member)
return {"message": "Face added successfully", "member": str(member.memberId)}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ฟังก์ชันสำหรับตรวจสอบใบหน้า
async def check_face(memberId: UUID, image_file, db: AsyncSession):
member = await db.get(Member, memberId)
if not member:
raise HTTPException(status_code=404, detail="Member not found")
if not member.faceEmbedding:
raise HTTPException(status_code=400, detail="No enrolled face found")
# บันทึกไฟล์ภาพชั่วคราว
temp_dir = Path("temp_images")
temp_dir.mkdir(exist_ok=True)
image_path = temp_dir / f"{uuid.uuid4()}.jpg"
with image_path.open("wb") as buffer:
shutil.copyfileobj(image_file.file, buffer)
try:
output = check_face_regonite(str(image_path), str(memberId))
if "Face match found!" in output:
return {"message": output}
else:
raise HTTPException(status_code=401, detail="Face does not match")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
image_path.unlink() # ลบไฟล์ชั่วคราว
async def check_face_emp(image_file, db: AsyncSession):
# บันทึกไฟล์ภาพชั่วคราว
temp_dir = Path("temp_images")
temp_dir.mkdir(exist_ok=True)
image_path = temp_dir / f"{uuid.uuid4()}.jpg"
with image_path.open("wb") as buffer:
shutil.copyfileobj(image_file.file, buffer)
try:
output = check_face_regonite_emp(str(image_path))
if output:
return {"message": output}
else:
raise HTTPException(status_code=401, detail="Face does not match")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
image_path.unlink() # ลบไฟล์ชั่วคราว
API-Fast/src/controllers/position_controller.py deleted 100644 → 0
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import joinedload
from ..models.position import Position
from ..models.company import Company
from fastapi import HTTPException, status
# ดึงข้อมูล position ทั้งหมด
async def get_all_positions(db: AsyncSession):
result = await db.execute(select(Position).options(joinedload(Position.company)))
return result.scalars().all()
# ดึงข้อมูล position ตาม ID
async def get_position_by_id(positionId: str, db: AsyncSession):
result = await db.execute(select(Position).where(Position.positionId == positionId).options(joinedload(Position.company)))
position = result.scalar_one_or_none()
if not position:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Position not found")
return position
# ดึงข้อมูล positions โดยใช้ companyId
async def get_positions_by_companyId(companyId: str, db: AsyncSession):
result = await db.execute(select(Position).where(Position.companyId == companyId))
return result.scalars().all()
# สร้าง position ใหม่
async def create_position(position_data: dict, db: AsyncSession):
# ตรวจสอบว่ามี companyId ที่อ้างอิงถึงอยู่หรือไม่
companyId = position_data["companyId"]
company = await db.get(Company, companyId)
if not company:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Company not found")
new_position = Position(**position_data)
db.add(new_position)
await db.commit()
await db.refresh(new_position)
return new_position
# อัปเดต position
async def update_position(positionId: str, position_data: dict, db: AsyncSession):
position = await db.get(Position, positionId)
if not position:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Position not found")
# อัปเดตข้อมูล position
for key, value in position_data.items():
setattr(position, key, value)
await db.commit()
await db.refresh(position)
return position
# ลบ position
async def delete_position(positionId: str, db: AsyncSession):
position = await db.get(Position, positionId)
if not position:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Position not found")
await db.delete(position)
await db.commit()
API-Fast/src/models/company.py deleted 100644 → 0
from sqlalchemy import Column, String, Integer, DateTime, Enum , Float
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from ..config.database import Base
import uuid
from sqlalchemy.orm import relationship
import enum
from datetime import datetime
class PublicType(enum.Enum):
PUBLIC = 1
PENDING = 0
class Company(Base):
__tablename__ = 'company'
companyId = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
companyName = Column(String, nullable=False, unique=True)
companyCode = Column(String, nullable=False, unique=True)
companyInfo = Column(String, nullable=False)
picture = Column(String)
address = Column(String, nullable=False)
latitude = Column(Float, nullable=False, default=0.00)
longitude = Column(Float, nullable=False, default=0.00)
status = Column(Enum(PublicType), nullable=False, default=PublicType.PUBLIC) # Default as Enum instance
createdAt = Column(DateTime, nullable=False, default=datetime.utcnow)
updatedAt = Column(DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
ownerName = Column(String, default="")
contact = Column(String, default="")
# ความสัมพันธ์
departments = relationship("Department", back_populates="company", cascade="all, delete")
positions = relationship("Position", back_populates="company", cascade="all, delete")
ProjectEmployees = relationship("ProjectEmployee", back_populates="company")
\ No newline at end of file
API-Fast/src/models/company_location.py deleted 100644 → 0
from sqlalchemy import Column, String, DateTime, Float, ForeignKey , Enum
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.sql import func
from ..config.database import Base
import uuid
import enum
class LocationType(enum.Enum):
PERMANENT = 0
TEMPORARY = 1
class CompanyLocation(Base):
__tablename__ = 'company_location'
locationId = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
companyId = Column(UUID(as_uuid=True), ForeignKey('company.companyId'), nullable=False)
latitude = Column(Float, nullable=False)
longitude = Column(Float, nullable=False)
locationName = Column(String, nullable=False)
radius = Column(Float, nullable=False ,default=0.5)
locationType = Column(Enum(LocationType), nullable=False, default=LocationType.PERMANENT)
startDate = Column(DateTime(timezone=True), server_default=func.now())
endDate = Column(DateTime(timezone=True), server_default=func.now())
createdAt = Column(DateTime(timezone=True), server_default=func.now())
updatedAt = Column(DateTime(timezone=True), onupdate=func.now())
API-Fast/src/models/position.py deleted 100644 → 0
from sqlalchemy import Column, String, ForeignKey , DateTime
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from ..config.database import Base
from sqlalchemy.sql import func
import uuid
class Position(Base):
__tablename__ = 'position'
positionId = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
thName = Column(String, nullable=False)
engName = Column(String, nullable=False)
companyId = Column(UUID(as_uuid=True), ForeignKey('company.companyId', ondelete="CASCADE"), nullable=False)
createdAt = Column(DateTime(timezone=True), server_default=func.now())
updatedAt = Column(DateTime(timezone=True), onupdate=func.now())
# ความสัมพันธ์กับ Company
company = relationship("Company", back_populates="positions")
ProjectEmployees = relationship("ProjectEmployee", back_populates="position")
API-Fast/src/models/project_employee.py deleted 100644 → 0
from sqlalchemy import Column, String, DateTime, Integer, ForeignKey , Enum , Float
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from ..config.database import Base
import uuid
import enum
class StatusType(enum.Enum):
ENABLE = 1
DISABLED = 0
class EmpType(enum.Enum):
PERMONTH = 1
PERDAY = 0
class CompanyRoleType(enum.Enum):
EMP = 0
ADMIN_COMPANY = 1
class CompanyEmployee(Base):
__tablename__ = 'company_employee'
company_employeeId = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
memberId = Column(UUID(as_uuid=True), ForeignKey('member.memberId'), nullable=False)
companyId = Column(UUID(as_uuid=True), ForeignKey('company.companyId'), nullable=False)
positionId = Column(UUID(as_uuid=True), ForeignKey('position.positionId'), nullable=False)
departmentId = Column(UUID(as_uuid=True), ForeignKey('department.departmentId'), nullable=False)
employeeId = Column(String, nullable=True, default="")
startDate = Column(DateTime(timezone=True), server_default=func.now())
empType = Column(Enum(EmpType), nullable=False, default=EmpType.PERMONTH) # Default as Enum instance
companyRoleType = Column(Enum(CompanyRoleType), nullable=False, default=CompanyRoleType.EMP) # Default as Enum instance
salary = Column(Float, nullable=False, default=0)
bossId = Column(String, nullable=True, default="")
createdAt = Column(DateTime(timezone=True), server_default=func.now())
updatedAt = Column(DateTime(timezone=True), onupdate=func.now())
# ความสัมพันธ์กับโมเดลอื่น ๆ
member = relationship("Member", back_populates="companyEmployees")
# boss = relationship("Member", back_populates="companyEmployees")
company = relationship("Company", back_populates="companyEmployees")
position = relationship("Position", back_populates="companyEmployees")
department = relationship("Department", back_populates="companyEmployees")
timestamps = relationship("EmployeeTimestamp", back_populates="company_employee",cascade="all, delete-orphan")
API-Fast/src/routes/company_location_routes.py deleted 100644 → 0
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from uuid import UUID
from ..config.database import get_db
from ..controllers.company_location_controller import (
get_all_locations,
get_location_by_id,
create_location,
update_location,
delete_location
)
from ..schemas.company_location_schema import CompanyLocationCreate, CompanyLocationUpdate, CompanyLocationResponse
router = APIRouter()
# ดึงข้อมูลสถานที่ทั้งหมดของบริษัท
@router.get("/company/{companyId}", response_model=List[CompanyLocationResponse])
async def get_all_locations_route(companyId: UUID, db: AsyncSession = Depends(get_db)):
return await get_all_locations(companyId, db)
# ดึงข้อมูลสถานที่ตาม locationId
@router.get("/company/{companyId}/{locationId}", response_model=CompanyLocationResponse)
async def get_location_by_id_route(companyId: UUID, locationId: UUID, db: AsyncSession = Depends(get_db)):
return await get_location_by_id(locationId, companyId, db)
# สร้างสถานที่ใหม่
@router.post("/company/{companyId}", response_model=CompanyLocationResponse)
async def create_location_route(companyId: UUID, location: CompanyLocationCreate, db: AsyncSession = Depends(get_db)):
return await create_location(location.dict(), companyId, db)
# อัปเดตข้อมูลสถานที่
@router.put("/company/{companyId}/{locationId}", response_model=CompanyLocationResponse)
async def update_location_route(companyId: UUID, locationId: UUID, location: CompanyLocationUpdate, db: AsyncSession = Depends(get_db)):
return await update_location(locationId, companyId, location.dict(exclude_unset=True), db)
# ลบสถานที่
@router.delete("/company/{companyId}/{locationId}")
async def delete_location_route(companyId: UUID, locationId: UUID, db: AsyncSession = Depends(get_db)):
await delete_location(locationId, companyId, db)
return {"detail": "Location deleted successfully"}
API-Fast/src/routes/company_routes.py deleted 100644 → 0
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from ..config.database import get_db
from ..controllers.company_controller import (
get_all_companies,
get_company_by_id,
create_company,
update_company,
delete_company
)
from ..schemas.company_schema import CompanyUpdate, CompanyResponse , CompanyBase
from typing import List
router = APIRouter()
@router.get("/", response_model=List[CompanyResponse])
async def get_companies_route(db: AsyncSession = Depends(get_db)):
return await get_all_companies(db)
@router.get("/{companyId}", response_model=CompanyResponse)
async def get_company_by_id_route(companyId: str, db: AsyncSession = Depends(get_db)):
return await get_company_by_id(companyId, db)
@router.post("/", response_model=CompanyResponse)
async def create_company_route(company: CompanyBase, db: AsyncSession = Depends(get_db)):
return await create_company(company.dict(), db)
@router.put("/{companyId}", response_model=CompanyResponse)
async def update_company_route(
companyId: str, company_update: CompanyUpdate, db: AsyncSession = Depends(get_db)
):
return await update_company(companyId, company_update.dict(exclude_unset=True), db)
@router.delete("/{companyId}")
async def delete_company_route(companyId: str, db: AsyncSession = Depends(get_db)):
return await delete_company(companyId, db)
API-Fast/src/routes/department_routes.py deleted 100644 → 0
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from ..config.database import get_db
from ..controllers.department_controller import (
get_all_departments,
get_department_by_id,
get_departments_by_companyId,
create_department,
update_department,
delete_department
)
from ..schemas.department_schema import DepartmentCreate, DepartmentUpdate, DepartmentResponse
router = APIRouter()
@router.get("/", response_model=List[DepartmentResponse])
async def get_departments_route(db: AsyncSession = Depends(get_db)):
return await get_all_departments(db)
@router.get("/{departmentId}", response_model=DepartmentResponse)
async def get_department_by_id_route(departmentId: str, db: AsyncSession = Depends(get_db)):
return await get_department_by_id(departmentId, db)
@router.get("/company/{companyId}", response_model=List[DepartmentResponse])
async def get_departments_by_companyId_route(companyId: str, db: AsyncSession = Depends(get_db)):
return await get_departments_by_companyId(companyId, db)
@router.post("/", response_model=DepartmentResponse)
async def create_department_route(department: DepartmentCreate, db: AsyncSession = Depends(get_db)):
return await create_department(department.dict(), db)
@router.put("/{departmentId}", response_model=DepartmentResponse)
async def update_department_route(departmentId: str, department: DepartmentUpdate, db: AsyncSession = Depends(get_db)):
return await update_department(departmentId, department.dict(exclude_unset=True), db)
@router.delete("/{departmentId}")
async def delete_department_route(departmentId: str, db: AsyncSession = Depends(get_db)):
await delete_department(departmentId, db)
return {"detail": "Department deleted successfully"}
API-Fast/src/routes/employee_routes.py deleted 100644 → 0
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from ..schemas.company_schema import CompanyBase, CompanyResponse
from ..config.database import get_db
from ..controllers.company_employee_controller import (
get_all_ProjectEmployees,
get_company_employee_by_id,
get_company_sub_employee_by_id,
update_company_employee,
delete_company_employee,
register_member,
get_companies_by_member,
get_companies_admin,
get_company_member_by_id
)
from ..schemas.company_employee_schema import ProjectEmployeeBase, ProjectEmployeePost, ProjectEmployeeUpdate, ProjectEmployeeResponse
from ..schemas.member_schema import MemberBase, MemberResponse
router = APIRouter()
# ดึงข้อมูลพนักงานในบริษัททั้งหมดตาม companyId
@router.get("/company/{companyId}", response_model=List[ProjectEmployeeResponse])
async def get_ProjectEmployees_route(companyId: str, db: AsyncSession = Depends(get_db)):
return await get_all_ProjectEmployees(companyId, db)
# ดึงข้อมูลพนักงานในบริษัทตาม company_employeeId และ companyId
@router.get("/company/{companyId}/{company_employeeId}", response_model=ProjectEmployeeResponse)
async def get_company_employee_by_id_route(companyId: str, company_employeeId: str, db: AsyncSession = Depends(get_db)):
return await get_company_employee_by_id(company_employeeId, companyId, db)
# ดึงข้อมูลพนักงานในบริษัทตาม memberId และ companyId
@router.get("/company/{companyId}/member/{memberId}", response_model=ProjectEmployeeResponse)
async def get_company_member_by_id_route(companyId: str, memberId: str, db: AsyncSession = Depends(get_db)):
return await get_company_member_by_id(memberId, companyId, db)
# ดึงลูกน้อง company_employeeId และ companyId
@router.get("/company/sub/{companyId}/{company_employeeId}", response_model=List[ProjectEmployeeResponse])
async def get_company_sub_employee_by_id_route(companyId: str, company_employeeId: str, db: AsyncSession = Depends(get_db)):
return await get_company_sub_employee_by_id(company_employeeId, companyId, db)
# เพิ่มสมาชิกใหม่หรือนำสมาชิกที่มีอยู่แล้วเข้าสู่บริษัท
@router.post("/company/{companyId}", response_model=ProjectEmployeeResponse)
async def register_member_route(
companyId: str,
member_data: ProjectEmployeePost,
db: AsyncSession = Depends(get_db)
):
return await register_member(member_data, companyId, db)
# แก้ไขข้อมูลพนักงานในบริษัทตาม company_employeeId และ companyId
@router.put("/company/{companyId}", response_model=ProjectEmployeeResponse)
async def update_company_employee_route(companyId: str, company_employee: ProjectEmployeeUpdate, db: AsyncSession = Depends(get_db)):
return await update_company_employee(company_employee, companyId,db)
# ลบพนักงานออกจากบริษัทตาม company_employeeId และ companyId
@router.delete("/company/{companyId}/{company_employeeId}")
async def delete_company_employee_route(companyId: str, company_employeeId: str, db: AsyncSession = Depends(get_db)):
await delete_company_employee(company_employeeId, companyId, db)
return {"detail": "Company Employee deleted successfully"}
@router.get("/member/{memberId}/companies", response_model=List[CompanyResponse])
async def get_companies_by_member_route(memberId: str, db: AsyncSession = Depends(get_db)):
return await get_companies_by_member(memberId, db)
@router.get("/member/{memberId}/companies-admin", response_model=List[CompanyResponse])
async def get_companies_by_admin_route(memberId: str, db: AsyncSession = Depends(get_db)):
return await get_companies_admin(memberId, db)
API-Fast/src/routes/employee_timestamp_routes.py deleted 100644 → 0
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Optional
from uuid import UUID
from ..models.employee_timestamp import TimestampType
from ..config.database import get_db
from ..controllers.employee_timestamp_controller import (
get_all_timestamps,
get_timestamp_by_id,
create_timestamp,
update_timestamp,
delete_timestamp,
search_timestamps
)
from ..schemas.employee_timestamp_schema import (
EmployeeTimestampCreate,
EmployeeTimestampUpdate,
EmployeeTimestampResponse,
EmployeeTimestampBase,
)
router = APIRouter()
# ดึงข้อมูลการลงเวลาทั้งหมดของพนักงานในบริษัท
@router.get("/company/{companyId}", response_model=List[EmployeeTimestampResponse])
async def get_all_timestamps_route(companyId: UUID, db: AsyncSession = Depends(get_db)):
return await get_all_timestamps(companyId, db)
# ดึงข้อมูลการลงเวลาของพนักงานจาก timestampId
@router.get("/company/{companyId}/{timestampId}", response_model=EmployeeTimestampResponse)
async def get_timestamp_by_id_route(companyId: UUID, timestampId: UUID, db: AsyncSession = Depends(get_db)):
return await get_timestamp_by_id(timestampId, companyId, db)
# สร้างข้อมูลการลงเวลาของพนักงานใหม่
@router.post("/company/{companyId}", response_model=EmployeeTimestampResponse)
async def create_timestamp_route(companyId: UUID, timestamp: EmployeeTimestampCreate, db: AsyncSession = Depends(get_db)):
return await create_timestamp(timestamp.dict(), companyId, db)
# อัปเดตข้อมูลการลงเวลาของพนักงาน
@router.put("/company/{companyId}/{timestampId}", response_model=EmployeeTimestampBase)
async def update_timestamp_route(companyId: UUID, timestampId: UUID, timestamp: EmployeeTimestampUpdate, db: AsyncSession = Depends(get_db)):
return await update_timestamp(timestampId, companyId, timestamp, db)
# ลบข้อมูลการลงเวลาของพนักงาน
@router.delete("/company/{companyId}/{timestampId}")
async def delete_timestamp_route(companyId: UUID, timestampId: UUID, db: AsyncSession = Depends(get_db)):
await delete_timestamp(timestampId, companyId, db)
return {"detail": "Timestamp deleted successfully"}
# ค้นหา timestamps ทั้งหมดของพนักงานจาก employeeId และเงื่อนไขอื่น ๆ
@router.get("/search/company/{companyId}", response_model=List[EmployeeTimestampResponse])
async def search_timestamps_route(
companyId: str,
company_employeeId: Optional[str] = Query(None),
startDate: Optional[datetime] = Query(None),
endDate: Optional[datetime] = Query(None),
timestampType: Optional[str] = Query(None),
db: AsyncSession = Depends(get_db)
):
return await search_timestamps(companyId, company_employeeId, startDate, endDate, timestampType, db)
API-Fast/src/routes/face_routes.py deleted 100644 → 0
from fastapi import APIRouter, Depends, UploadFile, File, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from ..config.database import get_db
from ..controllers.face_controller import add_face, check_face , check_face_emp
from uuid import UUID
router = APIRouter()
# เส้นทาง API สำหรับเพิ่มใบหน้า
@router.post("/members/{memberId}/add-face")
async def add_face_route(memberId: UUID, file: UploadFile = File(...), db: AsyncSession = Depends(get_db)):
return await add_face(memberId, file, db)
# เส้นทาง API สำหรับตรวจสอบใบหน้า
@router.post("/members/{memberId}/check-face")
async def check_face_route(memberId: UUID, file: UploadFile = File(...), db: AsyncSession = Depends(get_db)):
return await check_face(memberId, file, db)
# เส้นทาง API สำหรับตรวจสอบใบหน้า
@router.post("/members/check-face-emp")
async def check_face_emp_route(file: UploadFile = File(...), db: AsyncSession = Depends(get_db)):
return await check_face_emp(file, db)
\ No newline at end of file
API-Fast/src/routes/position_routes.py deleted 100644 → 0
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from ..config.database import get_db
from ..controllers.position_controller import (
get_all_positions,
get_position_by_id,
get_positions_by_companyId,
create_position,
update_position,
delete_position
)
from ..schemas.position_schema import PositionCreate, PositionUpdate, PositionResponse
router = APIRouter()
@router.get("/", response_model=List[PositionResponse])
async def get_positions_route(db: AsyncSession = Depends(get_db)):
return await get_all_positions(db)
@router.get("/{positionId}", response_model=PositionResponse)
async def get_position_by_id_route(positionId: str, db: AsyncSession = Depends(get_db)):
return await get_position_by_id(positionId, db)
@router.get("/company/{companyId}", response_model=List[PositionResponse])
async def get_positions_by_companyId_route(companyId: str, db: AsyncSession = Depends(get_db)):
return await get_positions_by_companyId(companyId, db)
@router.post("/", response_model=PositionResponse)
async def create_position_route(position: PositionCreate, db: AsyncSession = Depends(get_db)):
return await create_position(position.dict(), db)
@router.put("/{positionId}", response_model=PositionResponse)
async def update_position_route(positionId: str, position: PositionUpdate, db: AsyncSession = Depends(get_db)):
return await update_position(positionId, position.dict(exclude_unset=True), db)
@router.delete("/{positionId}")
async def delete_position_route(positionId: str, db: AsyncSession = Depends(get_db)):
await delete_position(positionId, db)
return {"detail": "Position deleted successfully"}
API-Fast/src/schemas/company_employee_schema.py deleted 100644 → 0
from pydantic import BaseModel , EmailStr
from typing import Optional
from uuid import UUID
from datetime import date, datetime
from ..models.company_employee import CompanyRoleType, EmpType, StatusType
from ..models.member import RoleType
class ProjectEmployeeBase(BaseModel):
memberId: UUID
companyId: UUID
positionId: UUID
departmentId: UUID
employeeId: str
class MemberInput(BaseModel):
email: EmailStr
firstName: Optional[str] = None
lastName: Optional[str] = None
picture: Optional[str] = None
class PositionInput(BaseModel):
positionId: UUID
thName: str
engName: str
class DepartmentInput(BaseModel):
departmentId: UUID
thName: str
engName: str
class ProjectEmployeePost(BaseModel):
member: MemberInput
position: Optional[PositionInput] = None
department: Optional[DepartmentInput] = None
employeeId: Optional[str] = None
salary: Optional[float] = None
bossId: Optional[str] = None
companyRoleType: CompanyRoleType
empType: EmpType
startDate: datetime
class ProjectEmployeeCreate(ProjectEmployeeBase):
pass
class ProjectEmployeeUpdate(BaseModel):
member: MemberInput
position: Optional[PositionInput] = None
department: Optional[DepartmentInput] = None
employeeId: Optional[str] = None
salary: Optional[float] = None
bossId: Optional[str] = None,
company_employeeId: UUID
companyRoleType: CompanyRoleType
empType: EmpType
startDate: datetime
class MemberResponse(BaseModel):
memberId: UUID
email: str
firstName: Optional[str]
lastName: Optional[str]
picture: Optional[str]
class PositionResponse(BaseModel):
positionId: UUID
thName: str
engName: str
class DepartmentResponse(BaseModel):
departmentId: UUID
thName: str
engName: str
class ProjectEmployeeResponse(BaseModel):
member: MemberResponse
companyId: UUID
position: Optional[PositionResponse]
department: Optional[DepartmentResponse]
company_employeeId: UUID
employeeId: str
salary: float
bossId: str
companyRoleType: CompanyRoleType
empType: EmpType
startDate: datetime
class Config:
from_attributes = True
\ No newline at end of file
API-Fast/src/schemas/company_schema.py deleted 100644 → 0
from pydantic import BaseModel
from typing import Optional
from uuid import UUID
from datetime import datetime
from ..models.company import PublicType
class CompanyBase(BaseModel):
companyName: str
companyCode: str
companyInfo: str
address: str
latitude: float
longitude: float
picture: str
status: PublicType
ownerName: str
contact: str
class CompanyUpdate(BaseModel):
companyName: Optional[str] = None
companyCode: Optional[str] = None
companyInfo: Optional[str] = None
address: Optional[str] = None
latitude: Optional[float] = None
longitude: Optional[float] = None
picture: Optional[str] = None
status: Optional[PublicType] = None
ownerName: Optional[str] = None
contact: Optional[str] = None
class CompanyResponse(CompanyBase):
companyId: UUID
createdAt: datetime
updatedAt: datetime
class Config:
from_attributes = True
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment