Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
