Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import time, datetime
- import json
- from flask import Flask, redirect, request, render_template
- from flask.json import jsonify
- from flask_sqlalchemy import SQLAlchemy
- from flask_restful import reqparse, abort, Api, Resource
- from flask_cors import CORS, cross_origin
- from uuid import uuid4
- app = Flask(__name__)
- CORS(app, resources=r'/amil/*')
- app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/amil.db'
- # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
- api = Api(app)
- CHEADER = {'Allow': 'POST,GET,PUT,DELETE'}, 200, {'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Methods': 'POST,PUT,GET', 'Access-Control-Allow-Headers': "Content-Type,Authorization"}
- # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
- @app.after_request
- def after_request(response):
- response.headers.add('Access-Control-Allow-Origin', '*')
- response.headers.add('Access-Control-Allow-Headers', 'Content-Type,Authorization')
- response.headers.add('Access-Control-Allow-Methods', 'GET,PUT,POST,DELETE')
- return response
- def check_authorization(request):
- if not request.get('Authorization', None) and request.get('Authorization', None) == 'aksjhd98jdasdo': #UUID
- abort(401, message="Authorization required")
- userparser = reqparse.RequestParser()
- userparser.add_argument('transaction_number', location='json', required=True)
- userparser.add_argument('cellphone_area', location='json', type=int, required=True)
- userparser.add_argument('cellphone_number', location='json', type=int, required=True)
- userparser.add_argument('cpf', location='json', type=int, required=True)
- userparser.add_argument('fullname', location='json', required=True)
- userparser.add_argument('birthdate', location='json', required=True)
- userparser.add_argument('gender_id', location='json', type=int, required=True)
- userparser.add_argument('state', location='json', required=True)
- userparser.add_argument('city', location='json', required=True)
- putuser = reqparse.RequestParser()
- putuser.add_argument('transaction_number', location='json', required=True)
- putuser.add_argument('cellphone_area', location='json', type=int, required=True)
- putuser.add_argument('cellphone_number', location='json', type=int, required=True)
- @app.route('/amil/user_data')
- def dummy():
- rows = User.query.all()
- return render_template("all.html", users=rows), 200
- def validate_create(args):
- try:
- assert len(args.get('transaction_number')) == 16
- assert len(str(args.get('cellphone_area'))) == 2
- assert len(str(args.get('cellphone_number'))) == 9
- assert len(str(args.get('cpf'))) == 11
- assert len(str(args.get('fullname'))) >=3 and len(str(args.get('fullname'))) <= 128
- try:
- assert datetime.datetime.strptime(args.get('birthdate'), '%Y-%m-%d')
- except ValueError:
- raise ValueError("Incorrect data format, should be YYYY-MM-DD")
- assert args.get('gender_id') >= 1 and args.get('gender_id') <=3
- assert len(str(args.get('state'))) == 2
- assert len(str(args.get('city'))) <= 128 and len(str(args.get('city'))) >= 3
- except AssertionError:
- raise AssertionError
- # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
- def validate_put(args):
- try:
- assert len(args.get('transaction_number')) == 16
- assert len(str(args.get('cellphone_area'))) == 2
- assert len(str(args.get('cellphone_number'))) == 9
- except AssertionError:
- raise AssertionError
- # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
- # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
- class UserController(Resource):
- def post(self):
- try:
- check_authorization(request.headers)
- args = userparser.parse_args()
- validate_create(args)
- y,m,d = args.get("birthdate").split("-");
- user = User.query.filter_by(transaction_number=args.get("transaction_number")).first()
- #criar usuario se jรก existir ativar
- if not user:
- userData = User(transaction_number=args.get("transaction_number"),
- cellphone_area=args.get("cellphone_area"),
- cellphone_number=args.get("cellphone_number"),
- cpf=args.get("cpf"),
- fullname=args.get("fullname"),
- birthdate=datetime.datetime(int(y), int(m), int(d)),
- gender_id=args.get("gender_id"),
- state=args.get("state"),
- city=args.get("city"),
- activated=1
- )
- db.session.add(userData)
- db.session.commit()
- msm_return = "novo"
- else:
- user.activated = 1
- user.updated_at = datetime.datetime.utcnow()
- user.deleted_at = None
- db.session.commit()
- msm_return = "ativado"
- return {"coisa": msm_return}, 200
- except AssertionError:
- abort(400, message='Field is not in required format')
- def delete(self):
- try:
- check_authorization(request.headers)
- args = putuser.parse_args()
- validate_put(args)
- #cancelado
- user = User.query.filter_by(transaction_number=args.get("transaction_number")).first()
- user.activated = 0
- user.updated_at = datetime.datetime.utcnow()
- user.deleted_at = datetime.datetime.utcnow()
- db.session.commit()
- return {"error":"", "message":"user has been canceled"}, 200
- except AssertionError:
- abort(400, message='Field is not in required format')
- def put(self):
- try:
- check_authorization(request.headers)
- args = putuser.parse_args()
- # validate_put(args)
- #inativar
- user = User.query.filter_by(transaction_number=args.get("transaction_number")).first()
- user.activated = 0
- user.updated_at = datetime.datetime.utcnow()
- db.session.commit()
- return {"error":"", "message":"user has been inatived"}, 200
- except AssertionError:
- abort(400, message='Field is not in required format')
- return data
- # - - - -- - - - - - - - - - - - - - -- - - - - - - - - - - - - - -- - - - - - - - - - -
- ## Database
- db = SQLAlchemy(app)
- # ------------------------------------
- class User(db.Model):
- id = db.Column(db.Integer, primary_key=True)
- transaction_number = db.Column(db.String(80), unique=True, nullable=False)
- cellphone_area = db.Column(db.Integer, nullable=False)
- cellphone_number = db.Column(db.Integer, nullable=False)
- cpf = db.Column(db.Integer, nullable=False)
- fullname = db.Column(db.String(128), nullable=False)
- birthdate = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow())
- gender_id = db.Column(db.Integer, nullable=False)
- state = db.Column(db.String(2), nullable=False)
- city = db.Column(db.String(128), nullable=False)
- activated = db.Column(db.Integer, nullable=True, default=0)
- created_at = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow())
- updated_at = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow())
- deleted_at = db.Column(db.DateTime, nullable=True)
- db.create_all()
- # - - - -- - - - - - - - - - - - - - -- - - - - - - - - - - - - - -- - - - - - - - - - -
- # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
- ##
- ## Actually setup the Api resource routing here
- ##
- api.add_resource(UserController, '/amil/user')
- # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
- if __name__ == '__main__':
- app.run(debug=True, host="0.0.0.0", threaded=True, port=8889)
Add Comment
Please, Sign In to add comment