How to use _api_id method in localstack

Best Python code snippet using localstack_python

api.py

Source:api.py

1# -*- coding: utf-8 -*-2import json3from odoo import _4from odoo.http import Controller, request, route5from ..tools import api_management, eval_request_params, RecordNotFoundError6def _get_model_obj(model, kwargs):7    user = request.env['res.users'].sudo().get_api_rest_user(8        request.httprequest.headers.get('x-api-key'))9    model_obj = request.env[model].with_user(user)10    # Used *in* of check kwargs to force remove11    # context when value {} is sending12    if 'context' in kwargs:13        model_obj = model_obj.with_context(14            **kwargs.get('context'))15        del kwargs['context']16    return model_obj17class RestApiDocs(Controller):18    @route("/api-docs/v<string:version>", auth="public", methods=["GET"])19    def api_docs(self, version=False, **kwargs):20        domain = [('name', '=', version)]21        version = request.env['api.rest.version'].search(domain, limit=1)22        if not version:23            return request.not_found()24        return request.render("smile_api_rest.openapi", {25            'url_swagger': version.url_swagger26        })27    @route("/api-docs/v<string:version>/swagger.json", auth="public",28           methods=["GET"])29    def api_json(self, version, **kwargs):30        domain = []31        if version:32            domain = [('name', '=', version)]33        version = request.env['api.rest.version'].search(domain, limit=1)34        if not version:35            return {'error': 'API Not found'}36        data = version.get_swagger_json()37        return request.make_response(json.dumps(data), headers=[38            ('Content-Type', 'application/json')39        ])40class RestApi(Controller):41    @route('/api/v<string:_api_version>/<string:_api_name>', auth='public',42           methods=["GET"], csrf=False)43    @api_management()44    def search_read(self, _api_version, _api_name, _api_path, **kwargs):45        eval_request_params(kwargs)46        _api_path._search_treatment_kwargs(kwargs)47        model_obj = _get_model_obj(_api_path.model, kwargs)48        return {49            'results': model_obj.search_read(**kwargs),50            'total': model_obj.search_count(kwargs.get('domain', [])),51            'offset': kwargs.get('offset', 0),52            'limit': kwargs.get('limit', 0),53            'version': _api_version,54        }55    @route('/api/v<string:_api_version>/<string:_api_name>/<int:_api_id>',56           auth='public', methods=["GET"], csrf=False)57    @api_management()58    def read(self, _api_version, _api_name, _api_id, _api_path, **kwargs):59        eval_request_params(kwargs)60        _api_path._read_treatment_kwargs(kwargs)61        model_obj = _get_model_obj(_api_path.model, kwargs)62        read_domain = _api_path._eval_domain(_api_path.filter_domain)63        read_domain += [('id', '=', _api_id)]64        record = model_obj.search(read_domain, limit=1)65        if not record:66            raise RecordNotFoundError(_('Record not found'))67        result = record.read(**kwargs)68        return result and result[0] or {}69    @route('/api/v<string:_api_version>/<string:_api_name>', auth='public',70           methods=["POST"], type='http', csrf=False)71    @api_management()72    def create(self, _api_version, _api_name, _api_path, **kwargs):73        eval_request_params(kwargs)74        model_obj = _get_model_obj(_api_path.model, kwargs)75        return model_obj.create(_api_path._post_treatment_values(kwargs)).id76    @route('/api/v<string:_api_version>/<string:_api_name>/<int:_api_id>',77           auth='public', methods=["PUT"], csrf=False)78    @api_management()79    def write(self, _api_version, _api_name, _api_id, _api_path, **kwargs):80        eval_request_params(kwargs)81        model_obj = _get_model_obj(_api_path.model, kwargs)82        update_domain = _api_path._eval_domain(_api_path.update_domain)83        update_domain += [('id', '=', _api_id)]84        record = model_obj.search(update_domain, limit=1)85        if not record:86            raise RecordNotFoundError(_('Record not found'))87        return record.write(_api_path._post_treatment_values(kwargs))88    @route('/api/v<string:_api_version>/<string:_api_name>/<int:_api_id>',89           auth='public', methods=["DELETE"], csrf=False)90    @api_management()91    def unlink(self, _api_version, _api_name, _api_id, _api_path, **kwargs):92        eval_request_params(kwargs)93        model_obj = _get_model_obj(_api_path.model, kwargs)94        unlink_domain = _api_path._eval_domain(_api_path.unlink_domain)95        unlink_domain += [('id', '=', _api_id)]96        record = model_obj.search(unlink_domain, limit=1)97        if not record:98            raise RecordNotFoundError(_('Record not found'))99        return record.unlink()100    @route([101        '/api/v<string:_api_version>/<string:_api_name>/<string:_api_method>',102        '/api/v<string:_api_version>/<string:_api_name>/<string:_api_method>/<int:_api_id>'103    ], auth='public', methods=["PUT"], csrf=False)104    @api_management()105    def custom_method(self, _api_version, _api_name, _api_method, _api_path,106                      _api_id=False, **kwargs):107        eval_request_params(kwargs)108        model_obj = _get_model_obj(_api_path.model, kwargs)109        if _api_id and _api_path.function_apply_on_record:110            function_domain = _api_path._eval_domain(_api_path.function_domain)111            function_domain += [('id', '=', _api_id)]112            record = model_obj.search(function_domain, limit=1)113            if not record:114                raise RecordNotFoundError(_('Record not found'))115        else:116            record = model_obj.browse()117        kwargs = _api_path._custom_treatment_values(kwargs)...

model.py

Source:model.py

1from flask import jsonify2from app import db3class Clone(db.Model):4    id = db.Column(db.Integer, primary_key=True)5    api_id = db.Column(db.String(255), unique=True)6    api_hash = db.Column(db.String(255), unique=True)7    username = db.Column(db.String(255))8    phone = db.Column(db.String(255))9    def __init__(self, _id, _api_id, _api_hash, _username, _phone) -> None:10        self.id = _id11        self.api_id = _api_id12        self.api_hash = _api_hash13        self.username = _username14        self.phone = _phone15    def addClone(_id, _api_id, _api_hash, _username, _phone):16        existed = Clone.checkExist(_api_id)17        if existed:18            return jsonify({"error": "Clone account existed"}), 40019        else:20            acc = Clone(_id, int(_api_id), _api_hash, _username, _phone)21            db.session.add(acc)22            db.session.commit()23            return jsonify({"status": "Add clone account successfully"}), 20024    def deleteClone(_api_id):25        Clone.query.filter_by(api_id=_api_id).delete()26        db.session.commit()27        return jsonify({"status": "Clone account was deleted"}), 20028    def deleteAllClone():29        Clone.query.delete()30        db.session.commit()31        return jsonify({"status": "All clone accounts were deleted"}), 20032    def editClone(_id, _api_id, _api_hash, _username, _phone):33        acc = Clone.query.filter_by(id=_id).first()34        acc.api_id = _api_id35        acc.api_hash = _api_hash36        acc.username = _username37        acc.phone = _phone38        db.session.commit()39        return jsonify({"status": "Update clone info successfully"}), 20040    def checkExist(_api_id):41        check = Clone.query.filter_by(api_id=_api_id).first()42        if check:43            return True44        else:45            return False46    def getAllClone():47        all = Clone.query.all()48        return all49    def getCloneWithID(_id):50        clone = Clone.query.filter_by(id=_id).first()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.