from array import array import datetime import base64 import json from odoo import http from odoo.http import request from odoo.tools.config import config from pytz import timezone import logging _logger = logging.getLogger(__name__) class Controller(http.Controller): def authenticate(self): wsgienv = request.httprequest.environ try: db = wsgienv['HTTP_DB'] username = wsgienv['HTTP_USERNAME'] password = wsgienv['HTTP_PASSWORD'] request.session.authenticate(db, username, password) return True except: try: authorization = wsgienv['HTTP_AUTHORIZATION'] except: authorization = None token = request.env['ir.config_parameter'].sudo().get_param('rest_api_token') or '' if authorization == token: request.session.authenticate(config.get('db_name'), 'it@fixcomart.co.id', 'Fixcomart378') return True return False def validate_request(self, rules: dict, kw: dict): validation = { 'status': True, 'reason': [] } for key in rules: values = rules[key] for value in values: if value == 'required' and not kw.get(key): validation['status'] = False validation['reason'].append(key + ' is ' + value) return validation def time_to_str(self, object, format): time = '' if isinstance(object, datetime.datetime): time = object.astimezone(timezone('Asia/Jakarta')).strftime(format) return time def response(self, data=[], code=200, description='OK'): response = { 'status': { 'code': code, 'description': description } } if code == 200: response.update({'result': data}) response = json.dumps(response) return request.make_response(response, [ ('Access-Control-Allow-Origin', '*'), ('Access-Control-Allow-Headers', '*'), ('Access-Control-Allow-Methods', '*'), ('Content-Type', 'application/json'), ]) def search_filter(self, model: str, kw: dict, query: array = []): """ To search data by default API Params if exist """ limit = kw.get('limit', 0) offset = kw.get('offset', 0) order = kw.get('order', '') return request.env[model].search(query, limit=int(limit), offset=int(offset), order=order) @http.route('/api/token', auth='public', methods=['GET', 'OPTIONS']) def get_api_token(self, **kw): return self.response(request.env['ir.config_parameter'].sudo().get_param('rest_api_token') or '') @http.route('/api/image///', auth='public', methods=['GET']) def get_image(self, model, field, id): model = request.env[model].sudo().search([('id', '=', id)], limit=1) image = model[field] if model[field] else '' return request.make_response(base64.b64decode(image), [('Content-Type', 'image/jpg')])