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()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!