How to use update_request_validator method in localstack
Best Python code snippet using localstack_python
request.py
Source:request.py 
1from flask import Blueprint, jsonify2from model import Request, Header, db, User, create_request_validator, update_request_validator, request_schema, requests_schema, Project3from decorators import token_required, load_data4request_api = Blueprint('request_api', __name__)5@request_api.route('/api/v1/projects/<project_id>/requests', methods=['GET'])6@token_required7def get_project_requests(current_user, project_id):8 """Devolver todas las requests."""9 project = Project.query.filter_by(project_id=project_id).first()10 if not project:11 return jsonify({'message': 'No project found!'}), 40412 if current_user != project.user:13 return jsonify({'message': 'You don\'t have permission to access these requests!'}), 40314 the_requests = project.requests15 output = requests_schema.dump(the_requests)16 data = output.data17 data.sort(key=lambda x: x["name"].lower())18 return jsonify({"requests": data})19@request_api.route('/api/v1/requests/<request_id>', methods=['GET'])20@token_required21def get_one_request(current_user, request_id):22 """Devuelve una request."""23 one_request = Request.query.filter_by(request_id=request_id).first()24 if not one_request:25 return jsonify({'message': 'No request found!'}), 40426 if current_user != one_request.project.user:27 return jsonify({'message': 'You don\'t have permission to access that request!'}), 40328 return jsonify({'request': request_schema.dump(one_request).data})29@request_api.route('/api/v1/projects/<project_id>/requests', methods=['POST'])30@token_required31@load_data32def create_request(data, current_user: User, project_id):33 """Crea una request."""34 project = Project.query.filter_by(project_id=project_id).first()35 if not project:36 return jsonify({'message': 'No project found!'}), 40437 if current_user != project.user:38 return jsonify({'message': 'You don\'t have permission to create a request in that project!'}), 40339 if create_request_validator.validate(data):40 for req in project.requests:41 if req.name == data['name']:42 return jsonify({'message': 'A request with that name already exists'}), 40943 one_request = Request(**data)44 db.session.add(one_request)45 project.requests.append(one_request)46 db.session.commit()47 return jsonify({'message': 'New request created!', 'request': request_schema.dump(one_request).data})48 else:49 return jsonify({'message': 'Request not created!', 'errors': create_request_validator.errors}), 40050@request_api.route('/api/v1/requests/<request_id>', methods=['DELETE'])51@token_required52def delete_request(current_user, request_id):53 """Devuelve una request."""54 one_request = Request.query.filter_by(request_id=request_id).first()55 if not one_request:56 return jsonify({'message': 'No request found!'}), 40457 if current_user != one_request.project.user:58 return jsonify({'message': 'You don\'t have permission to delete that request!'}), 40359 db.session.delete(one_request)60 db.session.commit()61 return jsonify({'message': 'The request has been deleted!'})62@request_api.route('/api/v1/requests/<request_id>', methods=['POST'])63@token_required64@load_data65def update_request(data, current_user, request_id):66 """Actualiza una request."""67 one_request = Request.query.filter_by(request_id=request_id).first()68 if not one_request:69 return jsonify({'message': 'No request found!'}), 40470 if current_user != one_request.project.user:71 return jsonify({'message': 'You don\'t have permission to update that request!'}), 40372 if update_request_validator.validate(data):73 # no actualizar si ya existe una request con ese nombre74 if Request.query.filter(Request.name == data['name'], Request.request_id != one_request.request_id).first():75 return jsonify({'message': 'A request with that name already exists'}), 40976 # eliminar las header anteriores77 for previous_header in one_request.headers:78 db.session.delete(previous_header)79 # rellenar datos de la header80 for key, value in data.items():81 # añadir directamente los campos que no sean las headers82 if key != "headers":83 setattr(one_request, key, value)84 # gestionar las headers85 else:86 # crear las nuevas headers87 for new_header_data in value:88 header = Header(**new_header_data)89 one_request.headers.append(header)90 db.session.commit()91 return jsonify({'message': 'Request updated!', 'request': request_schema.dump(one_request).data})92 else:...
model.py
Source:model.py
1from flask_sqlalchemy import SQLAlchemy2from flask_marshmallow import Marshmallow3from cerberus import Validator4import uuid5db = SQLAlchemy()6ma = Marshmallow()7def generate_uuid():8 """Genera un uuid"""9 return str(uuid.uuid4())10class User(db.Model):11 """Tabla usuarios de la base de datos."""12 user_id = db.Column(db.String(50), unique=True, primary_key=True, default=generate_uuid)13 email = db.Column(db.String(100), unique=True)14 password = db.Column(db.String(80), nullable=False)15 projects = db.relationship('Project', backref=db.backref('user'), cascade="all, delete-orphan")16_user_creation_schema = {17 'email': {'type': 'string', 'required': True, 'empty': False},18 'password': {'type': 'string', 'required': True, 'empty': False}19}20create_user_validator = Validator(_user_creation_schema)21class Project(db.Model):22 """Tabla proyectos de la base de datos."""23 project_id = db.Column(db.String(50), primary_key=True, default=generate_uuid)24 name = db.Column(db.String(100), nullable=False)25 user_id = db.Column(db.String(50), db.ForeignKey('user.user_id', ondelete='CASCADE'))26 requests = db.relationship('Request', backref=db.backref('project'), cascade="all, delete-orphan")27_project_creation_schema = {28 'name': {'type': 'string', 'required': True, 'empty': False},29}30create_project_validator = Validator(_project_creation_schema)31_project_update_schema = {32 'name': {'type': 'string', 'empty': False},33}34update_project_validator = Validator(_project_update_schema)35class Request(db.Model):36 """Tabla request de la base de datos"""37 request_id = db.Column(db.String(50), primary_key=True, default=generate_uuid)38 name = db.Column(db.String(100), nullable=False)39 url = db.Column(db.String(500))40 body = db.Column(db.Text)41 method = db.Column(db.String(50), default="GET")42 project_id = db.Column(db.String(50), db.ForeignKey('project.project_id', ondelete='CASCADE'))43 headers = db.relationship('Header', backref=db.backref('request'), cascade="all, delete-orphan")44_request_creation_schema = {45 'name': {'type': 'string', 'required': True, 'empty': False},46}47create_request_validator = Validator(_request_creation_schema)48_request_update_schema = {49 'name': {'type': 'string', 'empty': False},50 'url': {'type': 'string', 'empty': False},51 'body': {'type': 'string'},52 'method': {'type': 'string', 'empty': False},53 'headers': {'type': 'list',54 'schema': {'type': 'dict', 'schema': {55 'key': {'type': 'string', 'empty': False},56 'value': {'type': 'string', 'empty': False},57 }58 }59 }60}61update_request_validator = Validator(_request_update_schema)62class Header(db.Model):63 """Tabla header de la base de datos"""64 header_id = db.Column(db.String(50), primary_key=True, default=generate_uuid)65 key = db.Column(db.String(100), nullable=False)66 value = db.Column(db.String(100), nullable=False)67 request_id = db.Column(db.String(50), db.ForeignKey('request.request_id', ondelete='CASCADE'))68class UserSchema(ma.ModelSchema):69 """Esquema para la clase usuario."""70 class Meta:71 fields = ('user_id', 'email', 'projects')72 _links = ma.Hyperlinks(73 {"self": ma.URLFor("user_api.get_one_user", user_id="<user_id>"),74 "collection": ma.URLFor("user_api.get_all_users")75 }76 )77class ProjectSchema(ma.ModelSchema):78 """Esquema para la clase proyectos."""79 class Meta:80 model = Project81 _links = ma.Hyperlinks(82 {"self": ma.URLFor("project_api.get_one_project", project_id="<project_id>"),83 "collection": ma.URLFor("project_api.get_user_projects")}84 )85class RequestSchema(ma.ModelSchema):86 """Esquema para la clase proyectos."""87 class Meta:88 model = Request89 headers = ma.Nested('HeaderSchema', many=True, exclude=('header_id', 'request'))90 _links = ma.Hyperlinks(91 {"self": ma.URLFor("request_api.get_one_request", request_id="<request_id>"),92 "collection": ma.URLFor("request_api.get_project_requests", project_id="<project_id>")}93 )94class HeaderSchema(ma.ModelSchema):95 """Esquema para la clase Header."""96 class Meta:97 model = Header98user_schema = UserSchema()99users_schema = UserSchema(many=True)100project_schema = ProjectSchema()101projects_schema = ProjectSchema(many=True)102request_schema = RequestSchema()103requests_schema = RequestSchema(many=True)104header_schema = HeaderSchema()...
update_middleware.py
Source:update_middleware.py
1from constants.response_messages import VALID_UPDATE_COMMAND, INVALID_REQUEST_RESPONSE_CODE2from flask import abort3from flask import request4from functools import wraps5from validator.update_request_validator import UpdateRequestValidator6def validate_update_request(dbclient):7 def validate_update(f):8 @wraps(f)9 def wrap(*args, **kwargs):10 valid, message = UpdateRequestValidator(11 request, dbclient).isValid()12 if not valid:13 return '{}\n{}'.format(message, VALID_UPDATE_COMMAND), INVALID_REQUEST_RESPONSE_CODE14 else:15 return f(*args, **kwargs)16 return wrap...
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?
