-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapp.py
More file actions
101 lines (79 loc) · 2.99 KB
/
app.py
File metadata and controls
101 lines (79 loc) · 2.99 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
from flask import Flask, jsonify, request, send_file, render_template
from repository.database import db
from db_models.payment import Payment
from datetime import datetime, timedelta
from payments.pix import Pix
from flask_socketio import SocketIO
from docs.api.swagger_helper import add_swagger_routes
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
app.config['SECRET_KEY'] = 'SECRET_KEY_WEBSOCKET'
db.init_app(app)
socketio = SocketIO(app)
add_swagger_routes(app)
@app.route('/payments/pix', methods=['POST'])
def create_payment_pix():
data = request.get_json()
# Validation
if 'value' not in data:
return jsonify({"error": "Invalid value"}), 400
expiration_date = datetime.now() + timedelta(minutes=30)
new_payment = Payment(
value=data['value'],
expiration_date=expiration_date
)
pix_obj = Pix()
data_payment_pix = pix_obj.create_payment()
new_payment.bank_payment_id = data_payment_pix['bank_payment_id']
new_payment.qr_code = data_payment_pix['qr_code_path']
db.session.add(new_payment)
db.session.commit()
return jsonify({
"message": "The payment has been created",
"payment": new_payment.to_dict()
}), 201
@app.route('/payments/pix/qr_code/<file_name>', methods=['GET'])
def get_image(file_name):
return send_file(f'static/img/{file_name}.png', mimetype='image/png')
@app.route('/payments/pix/confirmation', methods=['GET'])
def pix_confirmation():
data = request.get_json()
# Validations
if 'bank_payment_id' not in data and "value" not in data:
return jsonify({"error": "Invalid payment data"}), 400
payment = Payment.query.filter_by(bank_payment_id=data.get('bank_payment_id')).first()
if not payment or payment.paid:
return jsonify({"message": "Payment Not Found"}), 404
if data.get("value") != payment.value:
return jsonify({"error": "Invalid payment data"}), 400
payment.paid = True
db.session.commit()
socketio.emit(f'payment-confirmed-{payment.id}')
return jsonify({"message": "The payment has been confirmated"}), 200
@app.route('/payments/pix/<int:payment_id>', methods=['GET'])
def payment_pix_page(payment_id):
payment = Payment.query.get(payment_id)
if not payment:
return render_template('404.html')
if payment.paid:
return render_template(
'confirmed_payment.html',
value=payment.value,
payment_id=payment.id
)
return render_template(
'payment.html',
payment_id=payment.id,
value=payment.value,
host='http://127.0.0.1:5000',
qr_code=payment.qr_code,
)
# Websockets
@socketio.on('connect')
def handle_connect():
print('Client connected to the Server')
@socketio.on('disconnect')
def handle_disconnect():
print('Client has been disconnected to the server')
if __name__ == "__main__":
socketio.run(app, debug=True)