devops_admin
/
odex25_standard
mirror of https://github.com/hydracp9/odex25_standard
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

get endpoin request employee and mission

This commit is contained in:
mohammed-alkhazrji 2025-05-18 11:30:02 +03:00
parent 99ca6a0d48
commit 0f02f5147e
4 changed files with 1650 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
odex25_mobile/odex_mobile/controllers/rest_api_v2/__init__.py
View File

@ -10,3 +10,4 @@ from . import pettie
from . import project_timesheet
from . import employee_other_request
from . import content_common
from . import official_mission

812
odex25_mobile/odex_mobile/controllers/rest_api_v2/employee_other_request.py
View File

@ -1,24 +1,36 @@
# -*- coding: utf-8 -*-
import werkzeug
from odoo import http, tools
from odoo.http import request, Response
from odoo.exceptions import UserError
import base64
from ...validator import validator
from ...http_helper import http_helper
from odoo.tools.translate import _
import json
import logging
import traceback
from odoo import http, _
from odoo.exceptions import UserError, AccessError, ValidationError
from odoo.http import request
from ...http_helper import http_helper
from ...validator import validator
from datetime import date, datetime
_logger = logging.getLogger(__name__)
def convert_dates_in_data(data):
if isinstance(data, dict):
for key, value in data.items():
data[key] = convert_dates_in_data(value)
return data
elif isinstance(data, list):
return [convert_dates_in_data(item) for item in data]
elif isinstance(data, (date, datetime)):
return data.isoformat()
else:
return data
class EmployeeOtherRequest(http.Controller):
def get_lable_selection(self, rec, field_name, state):
return dict(rec._fields[field_name]._description_selection(http.request.env)).get(state)
@http.route(['/rest_api/v2/employeeRequest/types', '/rest_api/v2/employeeRequest/types/<string:key>'], type='http', auth='none', csrf=False, methods=['GET'])
@http.route(['/rest_api/v2/employeeRequest/types', '/rest_api/v2/employeeRequest/types/<string:key>'], type='http',
auth='none', csrf=False, methods=['GET'])
def get_employee_other_request_type(self, key=None):
http_method, body, headers, token = http_helper.parse_request()
result = validator.verify_token(token)
@ -26,26 +38,45 @@ class EmployeeOtherRequest(http.Controller):
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400, message=_("You are not allowed to perform this operation. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"You are not allowed to perform this operation. please check with one of your team admins"),
success=False)
employee = http.request.env['hr.employee'].search(
[('user_id', '=', user.id)], limit=1)
if not employee:
return http_helper.response(code=400, message=_("You Have issue in your employee profile. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"You Have issue in your employee profile. please check with one of your team admins"), success=False)
try:
data = {"other_request_types": dict(http.request.env['employee.other.request']._fields['request_type']._description_selection(http.request.env)),
"salary_request_print_type": dict(http.request.env['employee.other.request']._fields['print_type']._description_selection(http.request.env)),
"salary_request_state": dict(http.request.env['employee.other.request']._fields['state']._description_selection(http.request.env)),
"certification_degree": dict(http.request.env['hr.certification']._fields['certification_degree']._description_selection(http.request.env)),
"qualification_degree": dict(http.request.env['hr.qualification']._fields['qualification_degree']._description_selection(http.request.env)),
"gender": dict(http.request.env['hr.employee.dependent']._fields['gender']._description_selection(http.request.env)),
"relation": dict(http.request.env['hr.employee.dependent']._fields['relation']._description_selection(http.request.env)),
data = {"other_request_types": dict(
http.request.env['employee.other.request']._fields['request_type']._description_selection(
http.request.env)),
"salary_request_print_type": dict(
http.request.env['employee.other.request']._fields['print_type']._description_selection(
http.request.env)),
"salary_request_state": dict(
http.request.env['employee.other.request']._fields['state']._description_selection(
http.request.env)),
"certification_degree": dict(
http.request.env['hr.certification']._fields['certification_degree']._description_selection(
http.request.env)),
"qualification_degree": dict(
http.request.env['hr.qualification']._fields['qualification_degree']._description_selection(
http.request.env)),
"gender": dict(http.request.env['hr.employee.dependent']._fields['gender']._description_selection(
http.request.env)),
"relation": dict(
http.request.env['hr.employee.dependent']._fields['relation']._description_selection(
http.request.env)),
"nationality": request.env['res.country'].sudo().search([]).read(['id', 'name']),
"uni_name_UniversityName": request.env['office.office'].sudo().search([]).read(['id', 'name']),
"col_name_College": request.env['hr.college'].sudo().search([]).read(['id', 'name']),
"hr_qualification_name": request.env['hr.qualification.name'].sudo().search([]).read(['id', 'name']),
"qualification_specification": request.env['qualification.specification'].sudo().search([('type', '=', 'qualification')]).read(['id', 'name']),
"certificate_specification": request.env['qualification.specification'].sudo().search([("type", "=", "certificate")]).read(['id', 'name']),
"hr_qualification_name": request.env['hr.qualification.name'].sudo().search([]).read(
['id', 'name']),
"qualification_specification": request.env['qualification.specification'].sudo().search(
[('type', '=', 'qualification')]).read(['id', 'name']),
"certificate_specification": request.env['qualification.specification'].sudo().search(
[("type", "=", "certificate")]).read(['id', 'name']),
"membership_type": request.env['membership.types'].sudo().search([]).read(['id', 'name']),
"membership_categorys": request.env['membership.categorys'].sudo().search([]).read(['id', 'name']),
}
@ -58,7 +89,8 @@ class EmployeeOtherRequest(http.Controller):
message = validator.get_server_error(e, user)
return http_helper.errcode(code=403, message=message)
@http.route(['/rest_api/v2/employeeRequests/', '/rest_api/v2/employeeRequests/<int:id>'], type='http', auth='none', csrf=False, methods=['GET'])
@http.route(['/rest_api/v2/employeeRequests/', '/rest_api/v2/employeeRequests/<int:id>'], type='http', auth='none',
csrf=False, methods=['GET'])
def get_employee_other_requests(self, id=None, approvel=None, page=None, **kw):
page = page if page else 1
page, offset, limit, prev = validator.get_page_pagination(page)
@ -68,21 +100,28 @@ class EmployeeOtherRequest(http.Controller):
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400, message=_("You are not allowed to perform this operation. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"You are not allowed to perform this operation. please check with one of your team admins"),
success=False)
employee = http.request.env['hr.employee'].search([('user_id', '=', user.id)], limit=1)
if not employee:
return http_helper.response(code=400, message=_("You Have issue in your employee profile. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"You Have issue in your employee profile. please check with one of your team admins"), success=False)
try:
if approvel:
domain = [('state', '!=', 'draft'),('employee_id', '!=', employee.id)]
emp_requests = http.request.env['employee.other.request'].search(domain, order='date desc', offset=offset, limit=limit)
domain = [('state', '!=', 'draft'), ('employee_id', '!=', employee.id)]
emp_requests = http.request.env['employee.other.request'].search(domain, order='date desc',
offset=offset, limit=limit)
count = http.request.env['employee.other.request'].search_count(domain)
else:
emp_requests = http.request.env['employee.other.request'].search([('employee_id', '=', employee.id)], order='date desc', offset=offset, limit = limit)
emp_requests = http.request.env['employee.other.request'].search([('employee_id', '=', employee.id)],
order='date desc', offset=offset,
limit=limit)
count = http.request.env['employee.other.request'].search_count([('employee_id', '=', employee.id)])
if id:
emp_requests = http.request.env['employee.other.request'].search([('id', '=', int(id))], order='date desc')
emp_requests = http.request.env['employee.other.request'].search([('id', '=', int(id))],
order='date desc')
count = http.request.env['employee.other.request'].search_count([('id', '=', int(id))])
employeeRequests = []
if emp_requests:
@ -96,7 +135,7 @@ class EmployeeOtherRequest(http.Controller):
'request_type': s.request_type,
'request_type_lable': self.get_lable_selection(s, 'request_type', s.request_type),
'state': s.state,
'state_lable': self.get_lable_selection(s, 'state', s.state),
'state_name': self.get_lable_selection(s, 'state', s.state),
}
employee_dependant = []
for dep in s.employee_dependant:
@ -104,7 +143,7 @@ class EmployeeOtherRequest(http.Controller):
'name': dep.name or '',
'age': dep.age or '',
'birthday': str(dep.birthday or ''),
'gender': dep.gender or '',
'gender': dep.gender or '',
'gender_lable': self.get_lable_selection(dep, 'gender', dep.gender),
'relation': dep.relation or '',
'relation_lable': self.get_lable_selection(dep, 'relation', dep.relation),
@ -137,8 +176,10 @@ class EmployeeOtherRequest(http.Controller):
'contact_email': qua.contact_email or '',
'country_name': qua.country_name.read(['id', 'name'])[0] or {},
'qualification_degree': qua.qualification_degree or '',
'qualification_degree_lable': self.get_lable_selection(qua, 'qualification_degree', qua.qualification_degree),
'qualification_specification_id': qua.qualification_specification_id.read(['id', 'name'])[0] or {},
'qualification_degree_lable': self.get_lable_selection(qua, 'qualification_degree',
qua.qualification_degree),
'qualification_specification_id': qua.qualification_specification_id.read(['id', 'name'])[
0] or {},
'qualification_id': qua.qualification_id.read(['id', 'name'])[0] or {},
# 'attachment': qua.attachment,
@ -153,8 +194,9 @@ class EmployeeOtherRequest(http.Controller):
'cer_name': cer.car_name or '',
'certification_specification': cer.certification_specification_id.name or '',
'issue_org': cer.issue_org or '',
'certification_degree': cer.certification_degree or '',
'certification_degree_lable': self.get_lable_selection(cer, 'certification_degree', cer.certification_degree),
'certification_degree': cer.certification_degree or '',
'certification_degree_lable': self.get_lable_selection(cer, 'certification_degree',
cer.certification_degree),
'issue_date': str(cer.issue_date or ''),
'exp_date': str(cer.exp_date or ''),
'country_id': cer.country_name.read(['id', 'name'])[0] or {},
@ -165,8 +207,8 @@ class EmployeeOtherRequest(http.Controller):
employeeRequests.append(value)
next = validator.get_page_pagination_next(page, count)
url = "/rest_api/v2/employeeRequests?approvel=%s&page=%s" % ( approvel, next) if next else False
prev_url = "/rest_api/v2/employeeRequests?approvel=%s&page=%s" % ( approvel, prev) if prev else False
url = "/rest_api/v2/employeeRequests?approvel=%s&page=%s" % (approvel, next) if next else False
prev_url = "/rest_api/v2/employeeRequests?approvel=%s&page=%s" % (approvel, prev) if prev else False
data = {'links': {'prev': prev_url, 'next': url, },
'count': count,
'results': {'employeeRequests': employeeRequests, }}
@ -189,29 +231,709 @@ class EmployeeOtherRequest(http.Controller):
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400, message=_("You are not allowed to perform this operation. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"You are not allowed to perform this operation. please check with one of your team admins"),
success=False)
employee = http.request.env['hr.employee'].search([('user_id', '=', user.id)], limit=1)
if not employee:
return http_helper.response(code=400, message=_("You Have issue in your employee profile. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"You Have issue in your employee profile. please check with one of your team admins"), success=False)
if not reportname:
return http_helper.response(code=400, message=_("please sent report name . please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"please sent report name . please check with one of your team admins"), success=False)
report = request.env['ir.actions.report']._get_report_from_name(reportname)
if docids:
docids = [int(i) for i in docids.split(',')]
else:
return http_helper.response(code=400, message=_("please sent id recrod print report. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"please sent id recrod print report. please check with one of your team admins"), success=False)
if report:
model = report.model_id.model or report.model
if len(request.env[model].search([('id', 'in', docids)])) < len(docids):
return http_helper.response(code=400, message=_("You Have issue in your data not found. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"You Have issue in your data not found. please check with one of your team admins"), success=False)
else:
return http_helper.response(code=400, message=_("You Have issue in your report not found. please check with one of your team admins"), success=False)
return http_helper.response(code=400, message=_(
"You Have issue in your report not found. please check with one of your team admins"), success=False)
try:
context = dict(request.env.context)
pdf = report.with_context(context)._render_qweb_pdf(docids, data=data)[0]
pdfhttpheaders = [('Content-Type', 'application/pdf'),('Content-Length', len(pdf))]
pdfhttpheaders = [('Content-Type', 'application/pdf'), ('Content-Length', len(pdf))]
return request.make_response(pdf, headers=pdfhttpheaders)
except Exception as e:
_logger.error(str(e))
message = validator.get_server_error(e, user)
return http_helper.errcode(code=403, message=message)
return http_helper.errcode(code=403, message=message)
class EmployeeOtherRequestController(http.Controller):
"""
Controller for Employee Other Requests (employee.other.request)
Endpoints:
3.1 Create Request (POST /rest_api/v2/employee_other_request)
3.2 Get All Requests (Paginate) (GET /rest_api/v2/employee_other_request)
3.3 Get Request by ID (GET /rest_api/v2/employee_request/<id>)
3.4 Update Request (PATCH /rest_api/v2/employee_other_request/<id>)
3.5 Delete Request (DELETE /rest_api/v2/employee_other_request/<id>)
"""
def get_lable_selection(self, rec, field_name, state):
return dict(rec._fields[field_name]._description_selection(http.request.env)).get(state)
# --------------------------------------------
# Utilities
# --------------------------------------------
def _prepare_employee_dependant_lines(self, dependant_list):
"""
Convert the list of employee_dependant dicts from the request
into Odoo One2many commands: (0, 0, {vals})
"""
commands = []
for dep in dependant_list or []:
commands.append((0, 0, {
"name": dep.get("name"),
"relation": dep.get("relation"),
"date_of_birth": dep.get("date_of_birth"),
}))
return commands
def _prepare_qualification_employee_lines(self, quals_list):
"""
Convert the list of qualification_employee dicts from the request
into One2many commands for hr.qualification model
"""
commands = []
for q in quals_list or []:
commands.append((0, 0, {
"qualification_name": q.get("qualification_name"),
"date_obtained": str(q.get("date_obtained","")),
"institute": q.get("institute"),
}))
return commands
def _prepare_certification_employee_lines(self, certs_list):
"""
Convert the list of certification_employee dicts from the request
into One2many commands for hr.certification model
"""
commands = []
for c in certs_list or []:
commands.append((0, 0, {
"certification_name": c.get("certification_name"),
"date_obtained":str( c.get("date_obtained","")),
"organization": c.get("organization"),
}))
return commands
def _get_request_return_data(self, req):
"""
Build a dictionary of fields to return for a single record
(similar to the detailed response in 3.3 Get Request by ID)
"""
res = {
"id": req.id,
"from_hr": req.from_hr,
"date": req.date,
"comment": req.comment or "",
"state": req.state,
'state_name': self.get_lable_selection(req, 'state', req.state),
"request_type": req.request_type,
"employee_id": req.employee_id.id if req.employee_id else None,
"employee_dependant": [],
"qualification_employee": [],
"certification_employee": [],
"create_insurance_request": req.create_insurance_request,
"print_type": req.print_type,
"destination": req.destination.id if req.destination else None,
"parent_request_id": req.parent_request_id.id if req.parent_request_id else None,
"company_id": req.company_id.id if req.company_id else None,
}
res = convert_dates_in_data(res)
# Build One2many lines for employee_dependant
for dep in req.employee_dependant:
res["employee_dependant"].append({
"name": dep.name,
"relation": dep.relation,
"date_of_birth": dep.date_of_birth,
})
# Build One2many lines for qualification_employee
for q in req.qualification_employee:
res["qualification_employee"].append({
"qualification_name": q.qualification_name,
"date_obtained": q.date_obtained,
"institute": q.institute,
})
res["qualification_employee"] = convert_dates_in_data(res["qualification_employee"])
# Build One2many lines for certification_employee
for c in req.certification_employee:
res["certification_employee"].append({
"certification_name": c.certification_name,
"date_obtained": c.date_obtained,
"organization": c.organization,
})
res["certification_employee"] = convert_dates_in_data(res["certification_employee"])
return res
# --------------------------------------------
# 3.1 Create Request (POST)
# --------------------------------------------
@http.route(['/rest_api/v2/employee_other_request'],
type='http', auth='none', csrf=False, methods=['POST'])
def create_request(self, **kw):
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(
code=400,
message=_("Authentication failed or user is not allowed."),
success=False
)
# 2) Validate/Parse Body
required_fields = ["date", "request_type", "employee_id", "company_id"]
missing = [f for f in required_fields if f not in body]
if missing:
return http_helper.response(
code=400,
message=_("Missing required fields: %s") % ", ".join(missing),
success=False
)
try:
with request.env.cr.savepoint():
# Prepare Odoo create vals
vals = {
"from_hr": body.get("from_hr", False),
"date": body["date"],
"comment": body.get("comment") or "",
"request_type": body["request_type"],
"employee_id": body["employee_id"],
"employee_dependant": self._prepare_employee_dependant_lines(body.get("employee_dependant", [])),
"qualification_employee": self._prepare_qualification_employee_lines(
body.get("qualification_employee", [])),
"certification_employee": self._prepare_certification_employee_lines(
body.get("certification_employee", [])),
"create_insurance_request": body.get("create_insurance_request", False),
"print_type": body.get("print_type", ""),
"destination": body.get("destination") or False,
"parent_request_id": body.get("parent_request_id") or False,
"company_id": body["company_id"],
}
# Create record
new_req = request.env["employee.other.request"].sudo().create(vals)
# Build success response
data = self._get_request_return_data(new_req)
return http_helper.response(
message=_("Request created successfully"),
data=data
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
_logger.error("Error creating request: %s", str(e))
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Unexpected error while creating request")
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# --------------------------------------------
# 3.2 Get All Requests (GET) with Pagination
# --------------------------------------------
@http.route(['/rest_api/v2/employee_other_request'],
type='http', auth='none', csrf=False, methods=['GET'])
def get_all_requests(self, **kw):
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(
code=400,
message=_("Authentication failed or user is not allowed."),
success=False
)
try:
# 2) Parse Query params
page = int(kw.get("page", 1))
limit = int(kw.get("limit", 10))
sort = kw.get("sort", "") # e.g. 'date' or '-state'
filters_str = kw.get("filters", "") # e.g. 'state=approved'
domain = []
if filters_str:
# Very naive filter parser: "state=approved;request_type=insurance"
for part in filters_str.split(";"):
if "=" in part:
k, v = part.split("=", 1)
if ',' in v:
domain.append((k.strip(), "in", v.split(',')))
else:
domain.append((k.strip(), "=", v.strip()))
domain.append(('request_type', 'in', ['salary_define', 'salary_fixing']))
offset = (page - 1) * limit if page > 0 else 0
order = False
if sort:
if sort.startswith("-"):
order = sort[1:] + " desc"
else:
order = sort + " asc"
RequestObj = request.env["employee.other.request"].with_user(user.id)
total_count = RequestObj.search_count(domain)
records = RequestObj.search(domain, offset=offset, limit=limit, order=order)
# Build minimal list
request_list = []
for r in records:
request_list.append(self._get_request_return_data(r))
request_list = convert_dates_in_data(request_list)
result_data = {
"page": page,
"limit": limit,
"total_records": total_count,
"requests": request_list,
}
return http_helper.response(
message=_("Requests retrieved successfully"),
data=result_data
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
_logger.error("Error getting requests: %s", str(e))
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Unexpected error while listing requests")
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# --------------------------------------------
# 3.3 Get Request by ID (GET)
# /rest_api/v2/employee_request/<id>
# --------------------------------------------
@http.route(['/rest_api/v2/employee_request/<int:req_id>'],
type='http', auth='none', csrf=False, methods=['GET'])
def get_request_by_id(self, req_id, **kw):
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(
code=400,
message=_("Authentication failed or user is not allowed."),
success=False
)
try:
req = request.env["employee.other.request"].sudo().search([("id", "=", req_id)], limit=1)
if not req:
return http_helper.response(code=404, message="Request not found", success=False)
data = self._get_request_return_data(req)
return http_helper.response(
message=_("Request retrieved successfully"),
data=data
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
_logger.error("Error getting request by ID: %s", str(e))
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Unexpected error while getting request by ID")
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# --------------------------------------------
# 3.4 Update Request (PATCH)
# /rest_api/v2/employee_other_request/<id>
# --------------------------------------------
@http.route(['/rest_api/v2/employee_other_request/<int:req_id>'],
type='http', auth='none', csrf=False, methods=['PATCH'])
def update_request(self, req_id, **kw):
"""
Allows partial update of fields present in the request body.
Example Body:
{
"state": "approved",
"comment": "Request approved by HR.",
"print_type": "no_salary"
}
"""
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(
code=400,
message=_("Authentication failed or user is not allowed."),
success=False
)
try:
with request.env.cr.savepoint():
req = request.env["employee.other.request"].sudo().search([("id", "=", req_id)], limit=1)
if not req:
return http_helper.response(code=404, message="Request not found", success=False)
# Build a vals dict for fields that appear in the body
updatable_fields = [
"from_hr", "date", "comment", "state", "request_type",
"employee_id", "create_insurance_request", "print_type",
"destination", "parent_request_id", "company_id"
]
vals = {}
for field_name in updatable_fields:
if field_name in body:
vals[field_name] = body[field_name]
# Handle One2many lines if needed
if "employee_dependant" in body:
# Clear existing lines, then add the new ones
vals["employee_dependant"] = [(5, 0, 0)]
vals["employee_dependant"] += self._prepare_employee_dependant_lines(body["employee_dependant"])
if "qualification_employee" in body:
vals["qualification_employee"] = [(5, 0, 0)]
vals["qualification_employee"] += self._prepare_qualification_employee_lines(
body["qualification_employee"])
if "certification_employee" in body:
vals["certification_employee"] = [(5, 0, 0)]
vals["certification_employee"] += self._prepare_certification_employee_lines(
body["certification_employee"])
req.write(vals)
updated_data = self._get_request_return_data(req)
return http_helper.response(
message=_("Request updated successfully"),
data=updated_data
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
_logger.error("Error updating request: %s", str(e))
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Unexpected error while updating request")
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# --------------------------------------------
# 3.5 Delete Request (DELETE)
# /rest_api/v2/employee_other_request/<int:req_id>
# --------------------------------------------
@http.route(['/rest_api/v2/employee_other_request/<int:req_id>'],
type='http', auth='none', csrf=False, methods=['DELETE'])
def delete_request(self, req_id, **kw):
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(
code=400,
message=_("Authentication failed or user is not allowed."),
success=False
)
try:
with request.env.cr.savepoint():
req = request.env["employee.other.request"].sudo().search([("id", "=", req_id)], limit=1)
if not req:
return http_helper.response(code=404, message="Request not found", success=False)
req.unlink()
return http_helper.response(
message=_("Request deleted successfully"),
data={"id": req_id}
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
_logger.error("Error deleting request: %s", str(e))
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Unexpected error while deleting request")
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# --------------------------------------------
# 4. Related Models (Read-Only)
# 4.1 Salary Destination
# 4.2 Get Qualifications
# 4.3 Get Employee Dependents
# 4.4 Get Certifications
# --------------------------------------------
class EmployeeOtherRequestRelatedModelsController(http.Controller):
"""
Minimal endpoints for related models used in employee.other.request
"""
def get_lable_selection(self, rec, field_name, state):
return dict(rec._fields[field_name]._description_selection(http.request.env)).get(state)
# 4.1 Salary Destination (salary.destination)
@http.route(['/rest_api/v2/salary_destination'], type='http', auth='none', csrf=False, methods=['GET'])
def get_salary_destination_list(self, **kw):
"""
GET /rest_api/v2/salary_destination
Return a list of salary.destination records
"""
http_method, body, headers, token = http_helper.parse_request()
# If this is publicly available, skip token check or do it similarly
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
try:
destinations = request.env['salary.destination'].sudo().search([])
data = []
for dest in destinations:
data.append({
"id": dest.id,
"name": dest.name,
"english_name": dest.english_name,
})
return http_helper.response(message="Salary Destination List", data=data)
except Exception as e:
_logger.exception("Error listing salary destinations")
return http_helper.response(code=400, message=str(e), success=False)
@http.route(['/api/salary_destination/<int:dest_id>'], type='http', auth='none', csrf=False, methods=['GET'])
def get_salary_destination_detail(self, dest_id, **kw):
"""
GET /api/salary_destination/<id>
Return the detail of a specific salary.destination record
(Note: endpoint path from your specification)
"""
# If you want token checks, add them
dest = request.env['salary.destination'].sudo().search([('id', '=', dest_id)], limit=1)
if not dest:
return http_helper.response(code=404, message="Salary destination not found", success=False)
data = {
"id": dest.id,
"name": dest.name,
"english_name": dest.english_name,
}
return http_helper.response(message="Salary Destination Detail", data=data)
# 4.2 Get Qualifications (Read-Only)
@http.route(['/rest_api/v2/qualifications'], type='http', auth='none', csrf=False, methods=['GET'])
def get_qualifications_list(self, **kw):
"""
GET /rest_api/v2/qualifications
Retrieve a paginated list of qualifications
Query Params:
page, limit, sort, filters
"""
http_method, body, headers, token = http_helper.parse_request()
# If needed, check token:
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
try:
page = int(kw.get("page", 1))
limit = int(kw.get("limit", 10))
sort = kw.get("sort", "") # e.g. '-comp_date'
filters_str = kw.get("filters", "")
domain = []
if filters_str:
for part in filters_str.split(";"):
if "=" in part:
k, v = part.split("=", 1)
domain.append((k.strip(), "=", v.strip()))
offset = (page - 1) * limit if page > 0 else 0
order = False
if sort:
if sort.startswith("-"):
order = sort[1:] + " desc"
else:
order = sort + " asc"
Qualification = request.env["hr.qualification"].sudo()
total_count = Qualification.search_count(domain)
records = Qualification.search(domain, offset=offset, limit=limit, order=order)
qual_list = []
for q in records:
qual_list.append({
"id": q.id,
"uni_name": q.uni_name.id if q.uni_name else None, # or q.uni_name.name?
"col_name": q.col_name.id if q.col_name else None,
"prg_status": q.prg_status or "",
"comp_date": q.comp_date or "",
"qualification_degree": q.qualification_degree or "",
"country_name": q.country_name.id if q.country_name else None,
"attachment": None,
})
qual_list = convert_dates_in_data(qual_list)
data = {
"page": page,
"limit": limit,
"total_records": total_count,
"qualifications": qual_list,
}
return http_helper.response(message="Qualifications retrieved", data=data)
except Exception as e:
_logger.exception("Error listing qualifications")
return http_helper.response(code=400, message=str(e), success=False)
# 4.3 Get Employee Dependents (Read-Only)
@http.route(['/rest_api/v2/employee_dependents'], type='http', auth='none', csrf=False, methods=['GET'])
def get_employee_dependents_list(self, **kw):
"""
GET /rest_api/v2/employee_dependents
Paginated list of dependents
Query Params:
page, limit, filters
"""
http_method, body, headers, token = http_helper.parse_request()
# If needed, check token:
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
try:
page = int(kw.get("page", 1))
limit = int(kw.get("limit", 10))
filters_str = kw.get("filters", "")
domain = []
if filters_str:
for part in filters_str.split(";"):
if "=" in part:
k, v = part.split("=", 1)
domain.append((k.strip(), "=", v.strip()))
offset = (page - 1) * limit if page > 0 else 0
DependentModel = request.env["hr.employee.dependent"].sudo()
total_count = DependentModel.search_count(domain)
records = DependentModel.search(domain, offset=offset, limit=limit)
dep_list = []
for d in records:
dep_list.append({
"id": d.id,
"name": d.name,
"birthday": d.birthday,
"age": d.age,
"gender": d.gender,
"relation": d.relation,
"nationality": d.nationality.name if d.nationality else "",
"passport_no": d.passport_no,
"passport_issue_date": d.passport_issue_date,
"passport_expire_date": d.passport_expire_date,
"remarks": d.remarks,
"has_ticket": d.has_ticket,
})
data = {
"page": page,
"limit": limit,
"total_records": total_count,
"dependents": dep_list,
}
return http_helper.response(message="Dependents retrieved", data=data)
except Exception as e:
_logger.exception("Error listing employee dependents")
return http_helper.response(code=400, message=str(e), success=False)
# 4.4 Get Certifications (Read-Only)
@http.route(['/rest_api/v2/certifications'], type='http', auth='none', csrf=False, methods=['GET'])
def get_certifications_list(self, **kw):
"""
GET /rest_api/v2/certifications
Paginated list of certifications
Query Params:
page, limit, filters
"""
http_method, body, headers, token = http_helper.parse_request()
# If needed, check token:
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
try:
page = int(kw.get("page", 1))
limit = int(kw.get("limit", 10))
filters_str = kw.get("filters", "")
domain = []
if filters_str:
for part in filters_str.split(";"):
if "=" in part:
k, v = part.split("=", 1)
domain.append((k.strip(), "=", v.strip()))
offset = (page - 1) * limit if page > 0 else 0
CertificationModel = request.env["hr.certification"].sudo()
total_count = CertificationModel.search_count(domain)
records = CertificationModel.search(domain, offset=offset, limit=limit)
cert_list = []
for c in records:
cert_list.append({
"id": c.id,
"car_name": c.car_name or "", # or maybe c.certification_name in your DB
"issue_org": c.issue_org or "",
"issue_date": c.issue_date or "",
"exp_date": c.exp_date or "",
"regis_no": c.regis_no or "",
"certification_degree": c.certification_degree or "",
"contact_name": c.contact_name or "",
"contact_phn": c.contact_phn or "",
"contact_email": c.contact_email or "",
"country_name": c.country_name.name if c.country_name else "",
"attachment": None,
})
cert_list = convert_dates_in_data(cert_list)
data = {
"page": page,
"limit": limit,
"total_records": total_count,
"certifications": cert_list,
}
return http_helper.response(message="Certifications retrieved", data=data)
except Exception as e:
_logger.exception("Error listing certifications")
return http_helper.response(code=400, message=str(e), success=False)

217
odex25_mobile/odex_mobile/controllers/rest_api_v2/employee_other_request_old.py Normal file
View File

@ -0,0 +1,217 @@
# -*- coding: utf-8 -*-
import werkzeug
from odoo import http, tools
from odoo.http import request, Response
from odoo.exceptions import UserError
import base64
from ...validator import validator
from ...http_helper import http_helper
from odoo.tools.translate import _
import json
import logging
import traceback
_logger = logging.getLogger(__name__)
class EmployeeOtherRequest(http.Controller):
def get_lable_selection(self, rec, field_name, state):
return dict(rec._fields[field_name]._description_selection(http.request.env)).get(state)
@http.route(['/old/rest_api/v2/employeeRequest/types', '/rest_api/v2/employeeRequest/types/<string:key>'], type='http', auth='none', csrf=False, methods=['GET'])
def get_employee_other_request_type(self, key=None):
http_method, body, headers, token = http_helper.parse_request()
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400, message=_("You are not allowed to perform this operation. please check with one of your team admins"), success=False)
employee = http.request.env['hr.employee'].search(
[('user_id', '=', user.id)], limit=1)
if not employee:
return http_helper.response(code=400, message=_("You Have issue in your employee profile. please check with one of your team admins"), success=False)
try:
data = {"other_request_types": dict(http.request.env['employee.other.request']._fields['request_type']._description_selection(http.request.env)),
"salary_request_print_type": dict(http.request.env['employee.other.request']._fields['print_type']._description_selection(http.request.env)),
"salary_request_state": dict(http.request.env['employee.other.request']._fields['state']._description_selection(http.request.env)),
"certification_degree": dict(http.request.env['hr.certification']._fields['certification_degree']._description_selection(http.request.env)),
"qualification_degree": dict(http.request.env['hr.qualification']._fields['qualification_degree']._description_selection(http.request.env)),
"gender": dict(http.request.env['hr.employee.dependent']._fields['gender']._description_selection(http.request.env)),
"relation": dict(http.request.env['hr.employee.dependent']._fields['relation']._description_selection(http.request.env)),
"nationality": request.env['res.country'].sudo().search([]).read(['id', 'name']),
"uni_name_UniversityName": request.env['office.office'].sudo().search([]).read(['id', 'name']),
"col_name_College": request.env['hr.college'].sudo().search([]).read(['id', 'name']),
"hr_qualification_name": request.env['hr.qualification.name'].sudo().search([]).read(['id', 'name']),
"qualification_specification": request.env['qualification.specification'].sudo().search([('type', '=', 'qualification')]).read(['id', 'name']),
"certificate_specification": request.env['qualification.specification'].sudo().search([("type", "=", "certificate")]).read(['id', 'name']),
"membership_type": request.env['membership.types'].sudo().search([]).read(['id', 'name']),
"membership_categorys": request.env['membership.categorys'].sudo().search([]).read(['id', 'name']),
}
if key:
data = {key: data[key]}
return http_helper.response(message="Data Found", data=data)
except Exception as e:
_logger.error(str(e))
_logger.error(traceback.format_exc())
message = validator.get_server_error(e, user)
return http_helper.errcode(code=403, message=message)
@http.route(['/old/rest_api/v2/employeeRequests/', '/rest_api/v2/employeeRequests/<int:id>'], type='http', auth='none', csrf=False, methods=['GET'])
def get_employee_other_requests(self, id=None, approvel=None, page=None, **kw):
page = page if page else 1
page, offset, limit, prev = validator.get_page_pagination(page)
http_method, body, headers, token = http_helper.parse_request()
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400, message=_("You are not allowed to perform this operation. please check with one of your team admins"), success=False)
employee = http.request.env['hr.employee'].search([('user_id', '=', user.id)], limit=1)
if not employee:
return http_helper.response(code=400, message=_("You Have issue in your employee profile. please check with one of your team admins"), success=False)
try:
if approvel:
domain = [('state', '!=', 'draft'),('employee_id', '!=', employee.id)]
emp_requests = http.request.env['employee.other.request'].search(domain, order='date desc', offset=offset, limit=limit)
count = http.request.env['employee.other.request'].search_count(domain)
else:
emp_requests = http.request.env['employee.other.request'].search([('employee_id', '=', employee.id)], order='date desc', offset=offset, limit = limit)
count = http.request.env['employee.other.request'].search_count([('employee_id', '=', employee.id)])
if id:
emp_requests = http.request.env['employee.other.request'].search([('id', '=', int(id))], order='date desc')
count = http.request.env['employee.other.request'].search_count([('id', '=', int(id))])
employeeRequests = []
if emp_requests:
for s in emp_requests:
value = {
'id': s.id,
'employee_id': s.employee_id.name,
'department_id': s.department_id.name,
'destination': s.destination,
'comment': s.comment,
'request_type': s.request_type,
'request_type_lable': self.get_lable_selection(s, 'request_type', s.request_type),
'state': s.state,
'state_lable': self.get_lable_selection(s, 'state', s.state),
}
employee_dependant = []
for dep in s.employee_dependant:
dep_val = {
'name': dep.name or '',
'age': dep.age or '',
'birthday': str(dep.birthday or ''),
'gender': dep.gender or '',
'gender_lable': self.get_lable_selection(dep, 'gender', dep.gender),
'relation': dep.relation or '',
'relation_lable': self.get_lable_selection(dep, 'relation', dep.relation),
'nationality': dep.nationality.read(['id', 'name'])[0] or {},
'passport_no': dep.passport_no or '',
'passport_issue_date': str(dep.passport_issue_date or ''),
'passport_expire_date': str(dep.passport_expire_date or ''),
# 'remarks': dep.remarks,
'degree_medical_insu': dep.degree_medical_insu or '',
'medical_insurance_num': dep.medical_insurance_num or '',
'identity_num': dep.identity_num or '',
'has_ticket': dep.has_ticket,
# 'attachment': dep.attachment,
}
employee_dependant.append(dep_val)
if s.employee_dependant:
value['employee_dependant'] = employee_dependant
qualification_employee = []
for qua in s.qualification_employee:
qua_val = {
'uni_name_UniversityName': qua.uni_name.read(['id', 'name'])[0] or {},
'col_name_CollegeName': qua.col_name.read(['id', 'name'])[0] or {},
'prg_status': qua.prg_status or '',
'comp_date': str(qua.comp_date or ''),
'end_date': str(qua.end_date or ''),
'degree': qua.degree or 0.0,
'contact_name': qua.contact_name or '',
'contact_phn': qua.contact_phn or '',
'contact_email': qua.contact_email or '',
'country_name': qua.country_name.read(['id', 'name'])[0] or {},
'qualification_degree': qua.qualification_degree or '',
'qualification_degree_lable': self.get_lable_selection(qua, 'qualification_degree', qua.qualification_degree),
'qualification_specification_id': qua.qualification_specification_id.read(['id', 'name'])[0] or {},
'qualification_id': qua.qualification_id.read(['id', 'name'])[0] or {},
# 'attachment': qua.attachment,
}
qualification_employee.append(qua_val)
if s.qualification_employee:
value['qualification_employee'] = qualification_employee
certification_employee = []
for cer in s.certification_employee:
cer_val = {
'id': cer.id,
'cer_name': cer.car_name or '',
'certification_specification': cer.certification_specification_id.name or '',
'issue_org': cer.issue_org or '',
'certification_degree': cer.certification_degree or '',
'certification_degree_lable': self.get_lable_selection(cer, 'certification_degree', cer.certification_degree),
'issue_date': str(cer.issue_date or ''),
'exp_date': str(cer.exp_date or ''),
'country_id': cer.country_name.read(['id', 'name'])[0] or {},
}
certification_employee.append(cer_val)
if s.certification_employee:
value['certification_employee'] = certification_employee
employeeRequests.append(value)
next = validator.get_page_pagination_next(page, count)
url = "/rest_api/v2/employeeRequests?approvel=%s&page=%s" % ( approvel, next) if next else False
prev_url = "/rest_api/v2/employeeRequests?approvel=%s&page=%s" % ( approvel, prev) if prev else False
data = {'links': {'prev': prev_url, 'next': url, },
'count': count,
'results': {'employeeRequests': employeeRequests, }}
return http_helper.response(message="Data Found", data=data)
except Exception as e:
_logger.error(str(e))
_logger.error(traceback.format_exc())
message = validator.get_server_error(e, user)
return http_helper.errcode(code=403, message=message)
@http.route([
'/old/rest_api/v2/report/<reportname>/<docids>',
'/rest_api/v2/report/',
], type='http', auth='none')
def report_routes(self, reportname=None, docids=None, **data):
http_method, body, headers, token = http_helper.parse_request()
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400, message=_("You are not allowed to perform this operation. please check with one of your team admins"), success=False)
employee = http.request.env['hr.employee'].search([('user_id', '=', user.id)], limit=1)
if not employee:
return http_helper.response(code=400, message=_("You Have issue in your employee profile. please check with one of your team admins"), success=False)
if not reportname:
return http_helper.response(code=400, message=_("please sent report name . please check with one of your team admins"), success=False)
report = request.env['ir.actions.report']._get_report_from_name(reportname)
if docids:
docids = [int(i) for i in docids.split(',')]
else:
return http_helper.response(code=400, message=_("please sent id recrod print report. please check with one of your team admins"), success=False)
if report:
model = report.model_id.model or report.model
if len(request.env[model].search([('id', 'in', docids)])) < len(docids):
return http_helper.response(code=400, message=_("You Have issue in your data not found. please check with one of your team admins"), success=False)
else:
return http_helper.response(code=400, message=_("You Have issue in your report not found. please check with one of your team admins"), success=False)
try:
context = dict(request.env.context)
pdf = report.with_context(context)._render_qweb_pdf(docids, data=data)[0]
pdfhttpheaders = [('Content-Type', 'application/pdf'),('Content-Length', len(pdf))]
return request.make_response(pdf, headers=pdfhttpheaders)
except Exception as e:
_logger.error(str(e))
message = validator.get_server_error(e, user)
return http_helper.errcode(code=403, message=message)

665
odex25_mobile/odex_mobile/controllers/rest_api_v2/official_mission.py Normal file
View File

@ -0,0 +1,665 @@
# -*- coding: utf-8 -*-
import logging
from odoo import http, _
from odoo.exceptions import UserError, AccessError, ValidationError
from odoo.http import request
from ...http_helper import http_helper
from ...validator import validator
from datetime import date, datetime
_logger = logging.getLogger(__name__)
def convert_dates_in_data(data):
if isinstance(data, dict):
for key, value in data.items():
data[key] = convert_dates_in_data(value)
return data
elif isinstance(data, list):
return [convert_dates_in_data(item) for item in data]
elif isinstance(data, (date, datetime)):
return data.isoformat()
else:
return data
class HrOfficialMissionController(http.Controller):
"""
Example Controller for HR Official Mission CRUD Endpoints:
1.1. Create Mission (POST /rest_api/v2/hr_official_mission)
1.2. Get All Missions (Paginate) (GET /rest_api/v2/hr_official_mission)
1.3. Get Mission by ID (GET /rest_api/v2/hr_official_mission/<id>)
1.4. Update Mission (PATCH /rest_api/v2/hr_official_mission/<id>)
1.5. Delete Mission (DELETE /rest_api/v2/hr_official_mission/<id>)
"""
def get_lable_selection(self, rec, field_name, state):
return dict(rec._fields[field_name]._description_selection(http.request.env)).get(state)
def _get_mission_return_data(self, mission):
"""
Build the dictionary of fields to return in mission responses.
This can be expanded according to your business needs.
"""
res = {
"id": mission.id,
"date": mission.date or False,
"move_type_selection": dict(
mission._fields['move_type']._description_selection(
http.request.env)),
"date_from": mission.date_from or False,
"date_to": mission.date_to or False,
"hour_from": mission.hour_from or 0.0,
"hour_to": mission.hour_to or 0.0,
"date_duration": mission.date_duration or 0.0,
"duration_type": mission.duration_type or "",
"hour_duration": mission.hour_duration or 0.0,
"balance": mission.balance or 0.0,
"early_exit": mission.early_exit or False,
"mission_purpose": mission.mission_purpose or "",
"state": mission.state or "",
'state_name': self.get_lable_selection(mission, 'state', mission.state),
"move_type": mission.move_type or "",
"move_type_name": self.get_lable_selection(mission, 'move_type', mission.move_type),
"department_id": mission.department_id.ids,
"employee_ids": [],
"approved_by": mission.approved_by.id if mission.approved_by else None,
"approved_by_name": mission.approved_by.name if mission.approved_by else None,
"refused_by": mission.refused_by.id if mission.refused_by else None,
"refused_by_name": mission.refused_by.name if mission.refused_by else None,
"mission_type": mission.mission_type.id if mission.mission_type else None,
"mission_type_name": mission.mission_type.name if mission.mission_type else None,
"country_id": mission.country_id.id if mission.country_id else None,
"country_name": mission.country_id.name if mission.country_id else None,
"ticket_insurance": mission.ticket_insurance or "",
"car_insurance": mission.car_insurance or "",
"self_car": mission.self_car or "",
"car_type": mission.car_type or "",
"rent_days": mission.rent_days or 0,
"max_rent": mission.max_rent or 0,
"visa": mission.visa or "",
"note": mission.note or "",
"course_name": mission.course_name.id if mission.course_name else None,
"course_name_name": mission.course_name.name if mission.course_name else None,
"process_type": mission.process_type or "",
"train_category": mission.train_category or "",
"partner_id": mission.partner_id.id if mission.partner_id else None,
"partner_name": mission.partner_id.name if mission.partner_id else None,
"destination": mission.destination.id if mission.destination else None,
"destination_name": mission.destination.name if mission.destination else None,
"issuing_ticket": mission.issuing_ticket or "",
"ticket_cash_request_type": mission.ticket_cash_request_type.id if mission.ticket_cash_request_type else None,
"ticket_cash_request_type_name": mission.ticket_cash_request_type.name if mission.ticket_cash_request_type else None,
"ticket_cash_request_for": mission.ticket_cash_request_for or "",
"Training_cost": mission.Training_cost or 0.0,
"appraisal_check": mission.appraisal_check or False,
"Tra_cost_invo_id": mission.Tra_cost_invo_id.id if mission.Tra_cost_invo_id else None,
"max_of_employee": mission.max_of_employee or 0,
"min_of_employee": mission.min_of_employee or 0,
"employee_id": mission.employee_id.id if mission.employee_id else None,
"employee_name": mission.employee_id.name if mission.employee_id else None,
"reference": mission.reference or "",
"company_id": mission.company_id.id if mission.company_id else None,
"company_name": mission.company_id.name if mission.company_id else None,
}
res = convert_dates_in_data(res)
# Build One2many list (hr.official.mission.employee)
employee_lines = []
for line in mission.employee_ids:
employee_lines.append({
"date_from": line.date_from or False,
"date_to": line.date_to or False,
"days": line.days or 0,
"hour_from": line.hour_from or 0,
"hour_to": line.hour_to or 0,
"hours": line.hours or 0,
"day_price": line.day_price or 0,
"hour_price": line.hour_price or 0,
"amount": line.amount or 0,
"fees_amount": line.fees_amount or 0,
"appraisal_id": line.appraisal_id.id if line.appraisal_id else None,
"appraisal_name": line.appraisal_id.name if line.appraisal_id else None,
"appraisal_result": line.appraisal_result.id if line.appraisal_result else None,
"employee_id": line.employee_id.id if line.employee_id else None,
"employee_name": line.employee_id.name if line.employee_id else None,
"account_move_id": line.account_move_id.id if line.account_move_id else None,
"advantage_id": line.advantage_id.id if line.advantage_id else None,
"train_cost_emp": line.train_cost_emp or 0,
"total_hours": line.total_hours or 0,
})
employee_lines = convert_dates_in_data(employee_lines)
res["employee_ids"] = employee_lines
return res
def _prepare_employee_ids(self, employee_ids_list):
"""
Convert the list of dictionaries (employee lines) from the request body
into the Odoo-compliant One2many / Many2many commands.
Each line => (0, 0, {...fields...}).
"""
commands = []
for line in employee_ids_list or []:
cmd_vals = {
"date_from": line.get("date_from"),
"date_to": line.get("date_to"),
"days": line.get("days", 0),
"hour_from": line.get("hour_from", 0),
"hour_to": line.get("hour_to", 0),
"hours": line.get("hours", 0),
"day_price": line.get("day_price", 0),
"hour_price": line.get("hour_price", 0),
"amount": line.get("amount", 0),
"fees_amount": line.get("fees_amount", 0),
"appraisal_id": line.get("appraisal_id") or False,
"appraisal_result": line.get("appraisal_result") or False,
"employee_id": line.get("employee_id") or False,
"account_move_id": line.get("account_move_id") or False,
"advantage_id": line.get("advantage_id") or False,
"train_cost_emp": line.get("train_cost_emp", 0),
"total_hours": line.get("total_hours", 0),
}
cmd_vals = convert_dates_in_data(cmd_vals)
commands.append((0, 0, cmd_vals))
return commands
# ----------------------------------------------------------
# 1.1 Create Mission (POST)
# ----------------------------------------------------------
@http.route(['/rest_api/v2/hr_official_mission'],
type='http', auth='none', csrf=False, methods=['POST'])
def create_mission(self, **kw):
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400,
message=_("Authentication failed or user is not allowed."),
success=False)
employee = request.env['hr.employee'].sudo().search([('user_id', '=', user.id)], limit=1)
# 2) Validate/Parse Body
# Required fields check, e.g. 'date_from', 'date_to', etc.
required_fields = ["date", "date_from", "date_to", "hour_from", "hour_to",
"mission_type"]
missing_fields = [f for f in required_fields if f not in body]
if missing_fields:
return http_helper.response(
code=400,
message=_("Missing required fields: %s") % ", ".join(missing_fields),
success=False
)
# 3) Create the record in Odoo
try:
with request.env.cr.savepoint():
# Convert department_id from a list to commands if needed
# department_ids = body.get("department_id", [])
# if department_ids and isinstance(department_ids, list):
# # For many2many, use [(6, 0, list_of_ids)]
# department_val = [(6, 0, department_ids)]
# else:
# department_val = False
# Convert employee_ids to one2many lines
# employee_lines = self._prepare_employee_ids(body.get("employee_ids", []))
# vals = {
# "date": body["date"],
# "date_from": body["date_from"],
# "date_to": body["date_to"],
# "hour_from": body["hour_from"],
# "hour_to": body["hour_to"],
# "date_duration": body.get("date_duration", 0),
# "hour_duration": body.get("hour_duration", 0),
# "balance": body.get("balance", 0),
# "early_exit": body.get("early_exit", False),
# "mission_purpose": body.get("mission_purpose", ""),
# # "move_type": body["move_type"],
# "department_id": department_val,
# "employee_ids": employee_lines,
# "approved_by": int(body.get("approved_by")) if body.get("approved_by") else False,
# "refused_by": int(body.get("refused_by")) if body.get("refused_by") else False,
# "mission_type": int(body["mission_type"]),
# "country_id": int(body.get("country_id")) if body.get("country_id") else False,
# "ticket_insurance": body.get("ticket_insurance", ""),
# "car_insurance": body.get("car_insurance", ""),
# "self_car": body.get("self_car", ""),
# "car_type": body.get("car_type", ""),
# "rent_days": body.get("rent_days", 0),
# "max_rent": body.get("max_rent", 0),
# "visa": body.get("visa", ""),
# "note": body.get("note", ""),
# "course_name": body.get("course_name") or False,
# "process_type": body.get("process_type", ""),
# "train_category": body.get("train_category", ""),
# "partner_id": int(body.get("partner_id")) if body.get("partner_id") else False,
# "issuing_ticket": body.get("issuing_ticket", ""),
# "ticket_cash_request_type": body.get("ticket_cash_request_type") or False,
# "ticket_cash_request_for": body.get("ticket_cash_request_for", ""),
# "Training_cost": body.get("Training_cost", 0.0),
# "appraisal_check": body.get("appraisal_check", False),
# "Tra_cost_invo_id": int(body.get("Tra_cost_invo_id")) if body.get("Tra_cost_invo_id") else False,
# "max_of_employee": body.get("max_of_employee", 0),
# "min_of_employee": body.get("min_of_employee", 0),
# "employee_id": employee.id,
# "reference": body.get("reference", ""),
# "company_id": 1,
# # attachments? see example below
# }
# Create the record
# Prepare the response
mission_type = kw.get('mission_type')
date_from = kw.get('date_from')
date_to = kw.get('date_to')
hour_from = kw.get('hour_from', 8)
hour_to = kw.get('hour_to', 16)
destination_type = kw.get('destination')
try:
hour_from = float(kw.get('hour_from') or 8)
except:
hour_from = 8.0
try:
hour_to = float(kw.get('hour_to') or 16)
except:
hour_to = 16.0
mission = request.env['hr.official.mission'].sudo().create({
'mission_type': int(mission_type),
'date_from': date_from,
'date_to': date_to,
'hour_from': hour_from,
'hour_to': hour_to,
'destination': int(destination_type),
'employee_id': employee.id,
'process_type': 'especially_hours',
})
request.env['hr.official.mission.employee'].sudo().create({
'employee_id': employee.id,
'official_mission_id': mission.id,
'date_from': date_from,
'date_to': date_to,
'hour_from': hour_from,
'hour_to': hour_to,
'days': mission.date_duration,
'hours': mission.hour_duration,
})
mis = request.env['hr.official.mission'].sudo().search([('id', '=', mission.id)])
data = self._get_mission_return_data(mis)
# Return success
return http_helper.response(
message=_("Mission created successfully"),
data=data
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Error creating mission: %s", e)
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# ----------------------------------------------------------
# 1.2 Get All Missions (List with Pagination)
# ----------------------------------------------------------
@http.route(['/rest_api/v2/hr_official_mission'],
type='http', auth='none', csrf=False, methods=['GET'])
def get_all_missions(self, **kw):
"""
Query parameters:
page (optional, default=1)
limit (optional, default=10)
sort (optional, e.g. 'date_from', '-state', etc.)
filters (optional, e.g. 'state=draft' or multiple conditions you parse)
"""
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400,
message=_("Authentication failed or user is not allowed."),
success=False)
try:
# 2) Build domain & pagination
page = int(kw.get('page', 1))
approvel = kw.get('approvel', 0)
sort = kw.get('sort') # e.g. 'date_from' or '-state'
filters_str = kw.get('filters')
page, offset, limit, prev = validator.get_page_pagination(page)
domain = []
if approvel:
domain.append(('state', '!=', 'draft'))
if filters_str:
# Very basic parse; adapt as you like.
parts = filters_str.split(';')
for p in parts:
if '=' in p:
key, val = p.split('=', 1)
domain.append((key.strip(), '=', val.strip()))
domain.append(('process_type', '=', 'especially_hours'))
offset = (page - 1) * limit if page > 0 else 0
order = "create_date desc"
if sort:
# if user specified '-field', interpret as descending
if sort.startswith('-'):
order = sort[1:] + " desc"
else:
order = sort + " asc"
# 3) Search & count
Mission = request.env['hr.official.mission'].with_user(user.id)
missions = Mission.search(domain, offset=offset, limit=limit, order=order)
all_missions = Mission.search_count(domain)
# 4) Build response data
data_list = []
for mis in missions:
print("process_type", mis.process_type)
data_list.append(self._get_mission_return_data(mis))
data_list = convert_dates_in_data(data_list)
next_page = validator.get_page_pagination_next(page, all_missions)
next_url = "/rest_api/v2/employee_other_request?approvel=%s&page=%s" % (
approvel, next_page) if next_page else False
prev_url = "/rest_api/v2/employee_other_request?approvel=%s&page=%s" % (approvel, prev) if prev else False
data = {
'links': {
'prev': prev_url,
'next': next_url,
},
'count': limit,
'results': {
'officialMission': data_list,
}
}
return http_helper.response(
message=_("All missions retrieved."),
data=data
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Error getting missions list: %s", e)
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# ----------------------------------------------------------
# 1.3 Get Mission by ID
# ----------------------------------------------------------
@http.route(['/rest_api/v2/hr_official_mission/<int:mission_id>'],
type='http', auth='none', csrf=False, methods=['GET'])
def get_mission_by_id(self, mission_id, **kw):
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400,
message=_("Authentication failed or user is not allowed."),
success=False)
try:
mission = request.env['hr.official.mission'].sudo().search([('id', '=', mission_id)], limit=1)
if not mission:
return http_helper.response(code=404, message="Mission not found", success=False)
data = self._get_mission_return_data(mission)
return http_helper.response(
message=_("Mission retrieved successfully"),
data=data
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Error getting mission by ID: %s", e)
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# ----------------------------------------------------------
# 1.4 Update Mission (PATCH)
# ----------------------------------------------------------
@http.route(['/rest_api/v2/hr_official_mission/<int:mission_id>'],
type='http', auth='none', csrf=False, methods=['PATCH'])
def update_mission(self, mission_id, **kw):
"""
Allows partial update of fields, only updates fields included in the request body.
Example body:
{
"date_to": "2025-02-10",
"hour_to": 18,
"balance": 120,
"mission_purpose": "Updated purpose for the mission."
}
"""
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400,
message=_("Authentication failed or user is not allowed."),
success=False)
try:
with request.env.cr.savepoint():
mission = request.env['hr.official.mission'].sudo().search([('id', '=', mission_id)], limit=1)
if not mission:
return http_helper.response(code=404, message="Mission not found", success=False)
# Build vals only from the body keys we actually have
vals = {}
updatable_fields = [
"date", "date_from", "date_to", "hour_from", "hour_to",
"date_duration", "hour_duration", "balance", "early_exit",
"mission_purpose", "state", "move_type", "approved_by",
"refused_by", "mission_type", "country_id", "ticket_insurance",
"car_insurance", "self_car", "car_type", "rent_days",
"max_rent", "visa", "note", "course_name", "process_type",
"train_category", "partner_id", "destination", "issuing_ticket",
"ticket_cash_request_type", "ticket_cash_request_for",
"Training_cost", "appraisal_check", "Tra_cost_invo_id",
"max_of_employee", "min_of_employee", "employee_id", "reference",
"company_id",
]
# If you need to handle department_id (many2many) or employee_ids (one2many) updates,
# you can handle them separately or the same way.
# e.g. if "department_id" in body: create the many2many commands.
# e.g. if "employee_ids" in body: parse them with _prepare_employee_ids.
for field_name in updatable_fields:
if field_name in body:
vals[field_name] = body[field_name]
# example for department_id (many2many):
if "department_id" in body:
dep_ids = body["department_id"]
if isinstance(dep_ids, list):
vals["department_id"] = [(6, 0, dep_ids)]
# example for employee_ids (one2many lines):
if "employee_ids" in body and isinstance(body["employee_ids"], list):
vals["employee_ids"] = [(5, 0, 0)] # remove existing lines
vals["employee_ids"] += self._prepare_employee_ids(body["employee_ids"])
mission.write(vals)
# Return the updated record
data = self._get_mission_return_data(mission)
return http_helper.response(
message=_("Mission updated successfully"),
data=data
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Error updating mission: %s", e)
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# ----------------------------------------------------------
# 1.5 Delete Mission (DELETE)
# ----------------------------------------------------------
@http.route(['/rest_api/v2/hr_official_mission/<int:mission_id>'],
type='http', auth='none', csrf=False, methods=['DELETE'])
def delete_mission(self, mission_id, **kw):
http_method, body, headers, token = http_helper.parse_request()
# 1) Check Token
result = validator.verify_token(token)
if not result['status']:
return http_helper.errcode(code=result['code'], message=result['message'])
user = validator.verify(token)
if not user:
return http_helper.response(code=400,
message=_("Authentication failed or user is not allowed."),
success=False)
try:
with request.env.cr.savepoint():
mission = request.env['hr.official.mission'].sudo().search([('id', '=', mission_id)], limit=1)
if not mission:
return http_helper.response(code=404, message="Mission not found", success=False)
mission.unlink()
return http_helper.response(
message=_("Mission deleted successfully"),
data={"id": mission_id}
)
except (UserError, AccessError, ValidationError) as e:
request.env.cr.rollback()
return http_helper.response(code=400, message=str(e), success=False)
except Exception as e:
request.env.cr.rollback()
_logger.exception("Error deleting mission: %s", e)
message = validator.get_server_error(e, user)
return http_helper.errcode(code=500, message=message)
# ------------------------------------------------------------------
# 2. Related Models (Read-Only)
# 2.1 Mission Type (hr.official.mission.type)
# 2.2 Mission Destination (mission.destination)
# 2.3 Employee Course Name (employee.course.name)
# ------------------------------------------------------------------
class HrOfficialMissionRelatedModelsController(http.Controller):
"""
Minimal read-only endpoints to retrieve valid IDs from related models
(mission_type, mission_destination, course_name, etc.)
"""
# 2.1 Mission Type
@http.route(['/api/mission_type'], type='http', auth='none', csrf=False, methods=['GET'])
def get_mission_type_list(self, **kw):
# No token check here if it is publicly available; otherwise, do similar token checks
domain = [('special_hours', '=', True)]
mission_types = request.env["hr.official.mission.type"].sudo().search(domain)
data = []
for mt in mission_types:
data.append({
"id": mt.id,
"name": mt.name,
"duration_type": mt.duration_type,
"maximum_days": mt.maximum_days,
"related_with_financial": mt.related_with_financial,
"day_price": mt.day_price,
"work_state": mt.work_state,
})
return http_helper.response(message="Mission Types", data=data)
@http.route(['/api/mission_type/<int:mt_id>'], type='http', auth='none', csrf=False, methods=['GET'])
def get_mission_type_detail(self, mt_id, **kw):
mt = request.env["hr.official.mission.type"].sudo().search([('id', '=', mt_id)], limit=1)
if not mt:
return http_helper.response(code=404, message="Mission Type not found", success=False)
data = {
"id": mt.id,
"name": mt.name,
"duration_type": mt.duration_type,
"maximum_days": mt.maximum_days,
"related_with_financial": mt.related_with_financial,
"day_price": mt.day_price,
"work_state": mt.work_state,
}
return http_helper.response(message="Mission Type Detail", data=data)
# 2.2 Mission Destination
@http.route(['/api/mission_destination'], type='http', auth='none', csrf=False, methods=['GET'])
def get_mission_destination_list(self, **kw):
destinations = request.env["mission.destination"].sudo().search([])
data = []
for dest in destinations:
data.append({
"id": dest.id,
"name": dest.name,
"code": dest.code,
"country_id": dest.country_id.id if dest.country_id else None,
})
return http_helper.response(message="Mission Destinations", data=data)
@http.route(['/api/mission_destination/<int:dest_id>'], type='http', auth='none', csrf=False, methods=['GET'])
def get_mission_destination_detail(self, dest_id, **kw):
dest = request.env["mission.destination"].sudo().search([('id', '=', dest_id)], limit=1)
if not dest:
return http_helper.response(code=404, message="Destination not found", success=False)
data = {
"id": dest.id,
"name": dest.name,
"code": dest.code,
"country_id": dest.country_id.id if dest.country_id else None,
}
return http_helper.response(message="Destination Detail", data=data)
# 2.3 Employee Course Name
@http.route(['/api/course_name'], type='http', auth='none', csrf=False, methods=['GET'])
def get_course_name_list(self, **kw):
courses = request.env["employee.course.name"].sudo().search([])
data = []
for c in courses:
data.append({
"id": c.id,
"name": c.name,
"code": c.code,
})
return http_helper.response(message="Course Names", data=data)
@http.route(['/api/course_name/<int:course_id>'], type='http', auth='none', csrf=False, methods=['GET'])
def get_course_name_detail(self, course_id, **kw):
course = request.env["employee.course.name"].sudo().search([('id', '=', course_id)], limit=1)
if not course:
return http_helper.response(code=404, message="Course not found", success=False)
data = {
"id": course.id,
"name": course.name,
"code": course.code,
}
return http_helper.response(message="Course Detail", data=data)