forked from scalablescripts/python-microservices
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
62 lines (44 loc) · 1.41 KB
/
Copy pathmain.py
File metadata and controls
62 lines (44 loc) · 1.41 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
from dataclasses import dataclass
from flask import Flask, jsonify, abort
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import UniqueConstraint
import requests
from producer import publish
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = 'mysql://root:root@db/main'
CORS(app)
db = SQLAlchemy(app)
@dataclass
class Product(db.Model):
id: int
title: str
image: str
id = db.Column(db.Integer, primary_key=True, autoincrement=False)
title = db.Column(db.String(200))
image = db.Column(db.String(200))
@dataclass
class ProductUser(db.Model):
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer)
product_id = db.Column(db.Integer)
UniqueConstraint('user_id', 'product_id', name='user_product_unique')
@app.route('/api/products')
def index():
return jsonify(Product.query.all())
@app.route('/api/products/<int:id>/like', methods=['POST'])
def like(id):
req = requests.get('http://docker.for.mac.localhost:8000/api/user')
json = req.json()
try:
productUser = ProductUser(user_id=json['id'], product_id=id)
db.session.add(productUser)
db.session.commit()
publish('product_liked', id)
except:
abort(400, 'You already liked this product')
return jsonify({
'message': 'success'
})
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0')