270 lines
8.6 KiB
Python
270 lines
8.6 KiB
Python
from dotenv import load_dotenv
|
|
import os
|
|
import logging
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from jose import jwt
|
|
from flask import Flask, request, jsonify
|
|
from flask_cors import CORS
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
import pymysql
|
|
from functools import wraps
|
|
from pymysql.cursors import DictCursor
|
|
import requests
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
app = Flask(__name__)
|
|
CORS(app)
|
|
app.secret_key = os.getenv('SECRET_KEY')
|
|
|
|
if not app.secret_key:
|
|
raise ValueError("SECRET_KEY must be set in the .env file")
|
|
|
|
TOKEN_VERIFICATION_URL = 'http://ww_user:5000/token'
|
|
|
|
db_config = {
|
|
'host': os.getenv('DB_HOST'),
|
|
'user': os.getenv('DB_USER'),
|
|
'password': os.getenv('DB_PASSWORD'),
|
|
'db': os.getenv('DB_NAME'),
|
|
'port': int(os.getenv('DB_PORT'))
|
|
}
|
|
|
|
for var in ['DB_HOST', 'DB_USER', 'DB_PASSWORD', 'DB_NAME', 'DB_PORT']:
|
|
if not os.getenv(var):
|
|
logging.warning(f"Environment variable {var} not found.")
|
|
|
|
def get_db_connection():
|
|
return pymysql.connect(**db_config)
|
|
|
|
def verify_token(f):
|
|
@wraps(f)
|
|
def wrapped(*args, **kwargs):
|
|
# Extract the token from the request headers
|
|
token = request.headers.get('Authorization', '').split()
|
|
|
|
if not token:
|
|
return jsonify({'error': 'No token provided'}), 401
|
|
|
|
token_type, token = token[0], token[1]
|
|
|
|
if token_type.lower() != 'bearer':
|
|
return jsonify({'error': 'Invalid token format'}), 400
|
|
|
|
try:
|
|
# Send the token to the verification service
|
|
response = requests.get(
|
|
TOKEN_VERIFICATION_URL,
|
|
headers={'Authorization': f'{token_type} {token}'}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return f(*args, **kwargs)
|
|
else:
|
|
return jsonify({'error': 'Invalid token'}), 401
|
|
|
|
except requests.RequestException as e:
|
|
# Handle network or service errors
|
|
return jsonify({'error': 'Token verification failed'}), 500
|
|
|
|
@app.route('/listing', methods=['POST'])
|
|
@verify_token
|
|
def add_listing():
|
|
"""Commit (save) a new listing"""
|
|
data = request.json
|
|
try:
|
|
connection = get_db_connection()
|
|
cursor = connection.cursor(DictCursor)
|
|
|
|
# Prepare SQL statement with placeholders for all required fields
|
|
sql = """
|
|
INSERT INTO Listing
|
|
(title, description, date, time, address, price, availability_status, user_id)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
"""
|
|
|
|
# Execute the statement with the provided data
|
|
cursor.execute(
|
|
sql,
|
|
(
|
|
data['title'],
|
|
data['description'],
|
|
data['date'],
|
|
data['time'],
|
|
data['address'],
|
|
data['price'],
|
|
data.get('availability_status', True),
|
|
data['user_id']
|
|
)
|
|
)
|
|
|
|
# Commit the transaction
|
|
connection.commit()
|
|
return jsonify({'message': 'Listing added successfully'}), 201
|
|
|
|
except pymysql.MySQLError as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
finally:
|
|
if connection:
|
|
connection.close()
|
|
|
|
@app.route('/listings', methods=['GET'])
|
|
def search_listings():
|
|
"""Search for listings based on filters"""
|
|
params = request.args.to_dict()
|
|
try:
|
|
connection = get_db_connection()
|
|
cursor = connection.cursor(DictCursor)
|
|
|
|
where_conditions = []
|
|
query_values = {}
|
|
|
|
if 'title' in params:
|
|
where_conditions.append("title LIKE %s")
|
|
query_values['title'] = f"%{params['title']}%"
|
|
if 'date' in params:
|
|
where_conditions.append("date = %s")
|
|
query_values['date'] = params['date']
|
|
if 'user_id' in params:
|
|
where_conditions.append("user_id = %s")
|
|
query_values['user_id'] = params['user_id']
|
|
|
|
sql = "SELECT * FROM Listing"
|
|
if where_conditions:
|
|
sql += f" WHERE {' AND '.join(where_conditions)}"
|
|
|
|
cursor.execute(sql, list(query_values.values()))
|
|
listings = cursor.fetchall()
|
|
|
|
# Convert date and time fields to strings for JSON serialization
|
|
for listing in listings:
|
|
if 'date' in listing:
|
|
# Convert DATE to string in 'YYYY-MM-DD' format
|
|
listing['date'] = listing['date'].isoformat()
|
|
if 'time' in listing:
|
|
# Convert TIME to string in 'HH:MM:SS' format
|
|
listing['time'] = (datetime.min + listing['time']).time().strftime('%H:%M:%S')
|
|
|
|
return jsonify(listings), 200
|
|
|
|
except pymysql.MySQLError as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
finally:
|
|
if connection:
|
|
connection.close()
|
|
|
|
@app.route('/listing/<int:listing_id>', methods=['GET'])
|
|
def get_listing(listing_id):
|
|
"""Get details of a specific listing"""
|
|
try:
|
|
connection = get_db_connection()
|
|
cursor = connection.cursor(DictCursor)
|
|
|
|
# Prepare SQL statement
|
|
sql = "SELECT * FROM Listing WHERE listing_id = %s"
|
|
|
|
cursor.execute(sql, (listing_id,))
|
|
listing = cursor.fetchone()
|
|
|
|
if not listing:
|
|
return jsonify({'message': 'Listing not found'}), 404
|
|
|
|
if 'date' in listing:
|
|
# Convert DATE to string in 'YYYY-MM-DD' format
|
|
listing['date'] = listing['date'].isoformat()
|
|
if 'time' in listing:
|
|
# Convert TIME to string in 'HH:MM:SS' format
|
|
listing['time'] = (datetime.min + listing['time']).time().strftime('%H:%M:%S')
|
|
|
|
return jsonify(listing), 200
|
|
|
|
except pymysql.MySQLError as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
finally:
|
|
if connection:
|
|
connection.close()
|
|
|
|
@app.route('/listing/<int:listing_id>', methods=['DELETE'])
|
|
@verify_token
|
|
def delete_listing(listing_id):
|
|
"""Delete a listing by ID"""
|
|
try:
|
|
connection = get_db_connection()
|
|
cursor = connection.cursor(DictCursor)
|
|
|
|
# Prepare SQL statement
|
|
sql = "DELETE FROM Listing WHERE listing_id = %s"
|
|
|
|
cursor.execute(sql, (listing_id,))
|
|
connection.commit()
|
|
return jsonify({'message': 'Listing deleted successfully'}), 200
|
|
|
|
except pymysql.MySQLError as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
finally:
|
|
if connection:
|
|
connection.close()
|
|
|
|
@app.route('/listing/<int:listing_id>', methods=['PUT'])
|
|
@verify_token
|
|
def change_listing(listing_id):
|
|
"""Update listing details"""
|
|
data = request.json
|
|
try:
|
|
connection = get_db_connection()
|
|
cursor = connection.cursor(DictCursor)
|
|
|
|
# Prepare SQL statement with only fields that are being updated
|
|
update_fields = []
|
|
update_values = []
|
|
|
|
if 'title' in data:
|
|
update_fields.append("title = %s")
|
|
update_values.append(data['title'])
|
|
if 'description' in data:
|
|
update_fields.append("description = %s")
|
|
update_values.append(data['description'])
|
|
if 'date' in data:
|
|
update_fields.append("date = %s")
|
|
update_values.append(data['date'])
|
|
if 'time' in data:
|
|
update_fields.append("time = %s")
|
|
update_values.append(data['time'])
|
|
if 'address' in data:
|
|
update_fields.append("address = %s")
|
|
update_values.append(data['address'])
|
|
if 'price' in data:
|
|
update_fields.append("price = %s")
|
|
update_values.append(data['price'])
|
|
if 'availability_status' in data:
|
|
update_fields.append("availability_status = %s")
|
|
update_values.append(data['availability_status'])
|
|
if 'user_id' in data:
|
|
update_fields.append("user_id = %s")
|
|
update_values.append(data['user_id'])
|
|
|
|
# Build the SQL query
|
|
sql = f"UPDATE Listing SET {', '.join(update_fields)} WHERE listing_id = %s"
|
|
update_values.append(listing_id)
|
|
|
|
if not update_fields:
|
|
return jsonify({'message': 'No fields to update'}), 400
|
|
|
|
cursor.execute(sql, tuple(update_values))
|
|
connection.commit()
|
|
return jsonify({'message': 'Listing updated successfully'}), 200
|
|
|
|
except pymysql.MySQLError as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
finally:
|
|
if connection:
|
|
connection.close()
|
|
|
|
@app.route('/')
|
|
def home():
|
|
return 'Hello v.01'
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True, host='0.0.0.0') |