I'm working on an application in Flask and am getting the error bellow
AssertionError: View function mapping is overwriting an existing endpoint function: index
for the following code:
import json
import random
import string
import os
import sys
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from app import app
from flask import jsonify, request
from database_helpers import query_db, insert_db
from werkzeug.security import generate_password_hash, check_password_hash
def decorator(func, request, check_token):
data = request.get_json(force=True)
try:
if check_token:
token = data.get('token', None)
if token:
user = query_db('SELECT * FROM Users WHERE token = ?', [token], one=True)
if user:
json, code = func(**data)
else:
json = jsonify(success=False, message='Invalid token')
code = 401
else:
json = jsonify(success=False, message='Misformatted data.')
code = 400
else:
json, code = func(**data)
except TypeError:
json = jsonify(success=False, message='Misformatted data.')
code = 400
return json, code
@app.route('/') <== Error given here
def index():
return "Hello world"
@app.route('/sign_in', methods=['POST'])
def sign_in_helper():
return decorator(sign_in, request, check_token=False)
def sign_in(email, password):
data = query_db('SELECT * FROM Users WHERE email = ?', [email], one=True)
if data and check_password_hash(data["password"], password):
token = token_creator()
insert_db('UPDATE Users SET token = ? WHERE email = ?', [token, email])
return jsonify(
succes=True,
message="Welcome",
data=json.dumps({'token': token}),
), 200
return jsonify(
success=False,
message="Username or password invalid"), 401
Similar questions on s.o. suggest that it is a naming issue. However, after renaming the index() function to any string, say in540604() the same error is given in the same place.
Traceback (most recent call last):
File "C:/Users/mg13/Desktop/School/WebPython/twidder/twidder/app/server.py", line 38, in <module>
@app.route('/')
File "C:\Python27\lib\site-packages\flask\app.py", line 1013, in decorator
self.add_url_rule(rule, endpoint, f, **options)
File "C:\Python27\lib\site-packages\flask\app.py", line 62, in wrapper_func
return f(self, *args, **kwargs)
File "C:\Python27\lib\site-packages\flask\app.py", line 984, in add_url_rule
'existing endpoint function: %s' % endpoint)
AssertionError: View function mapping is overwriting an existing endpoint function: in6537547
Process finished with exit code 1
Why?
Edit:
The entire code as well as the project directory tree are bellow:
server.py
import json
import random
import string
import os
import sys
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from app import app
from flask import jsonify, request
from database_helpers import query_db, insert_db
from werkzeug.security import generate_password_hash, check_password_hash
def decorator(func, request, check_token):
data = request.get_json(force=True)
try:
if check_token:
token = data.get('token', None)
if token:
user = query_db('SELECT * FROM Users WHERE token = ?', [token], one=True)
if user:
json, code = func(**data)
else:
json = jsonify(success=False, message='Invalid token')
code = 401
else:
json = jsonify(success=False, message='Misformatted data.')
code = 400
else:
json, code = func(**data)
except TypeError:
json = jsonify(success=False, message='Misformatted data.')
code = 400
return json, code
@app.route('/')
def index():
return "Hello world"
@app.route('/sign_in', methods=['POST'])
def sign_in_helper():
return decorator(sign_in, request, check_token=False)
def sign_in(email, password):
data = query_db('SELECT * FROM Users WHERE email = ?', [email], one=True)
if data and check_password_hash(data["password"], password):
token = token_creator()
insert_db('UPDATE Users SET token = ? WHERE email = ?', [token, email])
return jsonify(
succes=True,
message="Welcome",
data=json.dumps({'token': token}),
), 200
return jsonify(
success=False,
message="Username or password invalid"), 401
def token_creator():
data = query_db('SELECT token FROM Users', one=False)
while True:
token = ''.join(random.choice(string.ascii_letters + string.digits)
for x in range(32))
if token not in data:
return token
@app.route('/sign_up', methods=['POST'])
def sign_up_helper():
return decorator(sign_up, request, check_token=False)
def sign_up(email, password, firstname, familyname, gender, city, country):
"""
check if the email exists in the database
"""
data = query_db('SELECT * FROM Users WHERE email = ?', [email], one=True)
if data is not None:
return jsonify(
success=False,
message="Email already exists"), 400
insert_db('INSERT INTO Users (email, firstname, lastname, gender, city, country, password) \
VALUES (?, ?, ?, ?, ?, ?, ?)',
[email, firstname, familyname, gender, city,
country, generate_password_hash(password)])
return jsonify(
success=True,
message="Account created successfully"), 200
@app.route("/sign_out", methods=['PUT'])
def sign_out_helper():
return decorator(sign_out, request, check_token=True)
def sign_out(token):
query_db('UPDATE Users SET token = null WHERE token = ?', [token], one=True)
return jsonify(
success=True,
message="You are now signed out"), 200
@app.route("/change_password", methods=['PUT'])
def change_password_helper():
return decorator(change_password, request, check_token=True)
def change_password(token, old_password, new_password):
data = query_db('SELECT * FROM Users WHERE token = ?', [token], one=True)
if not check_password_hash(data["password"], old_password):
return jsonify(
success=False,
message="Incorect password"), 400
insert_db('UPDATE Users WHERE id=? SET password = ?', [data["id"],
new_password])
return jsonify(
success=True,
message="Password changed successfully"), 200
@app.route("/profile", methods=['GET'])
def get_user_data_by_token_helper():
return decorator(get_user_data_by_token, request, check_token=True)
def get_user_data_by_token(token):
data = query_db('SELECT * FROM Users WHERE token = ?', [token], one=True)
return jsonify(
success=True,
email=data["email"],
firstname=data["firstname"],
lastname=data["lastname"],
gender=data["gender"],
city=data["city"],
country=data["country"])
@app.route("/user", methods=['GET'])
def get_user_data_by_email_helper():
return decorator(get_user_data_by_email, request, check_token=True)
def get_user_data_by_email(token, email):
data = query_db('SELECT * FROM Users WHERE email = ? AND token = ?',
[email, token], one=True)
return jsonify(
success=True,
email=data["email"],
firstname=data["firstname"],
lastname=data["lastname"],
gender=data["gender"],
city=data["city"],
country=data["country"]), 200
@app.route("/profile_messages", methods=['GET'])
def get_user_messages_by_token_helper():
return decorator(get_user_messages_by_token, request, check_token=True)
def get_user_messages_by_token(token):
data = query_db('SELECT * FROM Users WHERE token = ?', [token], one=True)
data_messages = query_db('SELECT * FROM Messages WHERE sender = ?', [data["id"]], one=False)
return jsonify(
data=json.dumps(data_messages),
)
@app.route("/user_profile", methods=['GET'])
def get_user_messages_by_email_helper():
return decorator(get_user_messages_by_email, request, check_token=True)
def get_user_messages_by_email(token, email):
data = query_db('SELECT * FROM Users WHERE email= ?', [email], one=True)
data_messages = query_db('SELECT * FROM Messages WHERE sender = ?', [data["id"]], one=False)
return jsonify(
data=json.dumps(data_messages),
)
@app.route("/message", methods=['GET', 'POST'])
def post_message_helper():
return decorator(post_message, request, check_token=True)
def post_message(token, message, email):
data = query_db('SELECT * FROM Users WHERE email= ?', [email], one=True)
my_data = query_db('SELECT * FROM Users WHERE token= ?', [token], one=True)
if data is None:
return jsonify(
success=False,
message="Recipient does not exist"), 400
insert_db('INSERT INTO Messages (receiver, sender, message) \
VALUES (?, ?, ?, ?)', [data["id"], my_data["id"], message])
return jsonify(
success=True,
message="Message posted successfully"), 200
database_helpers.py
import os
import sys
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from app import app
from flask import g
import sqlite3
def init_db():
with app.app_context():
db = get_database()
with app.open_resource(app.config['SCHEMA'], mode='r') as f:
db.cursor().executescript(f.read())
db.commit()
def query_db(query, args=(), one=False):
cur = get_database().execute(query, args)
rv = cur.fetchall()
cur.close()
return (rv[0] if rv else None) if one else rv
def db_connect():
db = sqlite3.connect(app.config['DATABASE'])
db.row_factory = sqlite3.Row
return db
def get_database():
if not hasattr(g, 'db'):
g.db = db_connect()
return g.db
def insert_db(query, args=(), one=False):
cur = get_database().execute(query, args)
get_database().commit()
rv = cur.fetchall()
cur.close()
return (rv[0] if rv else None) if one else rv
__init__.py
import os
import sys
import platform
from flask import Flask, g
app = Flask(__name__)
__all__ = ["server", "database_helpers"]
if platform.system() == 'Windows':
app.config.update(dict(
DATABASE=os.path.join(app.root_path, 'database\database.db'),
SCHEMA=os.path.join(app.root_path, 'database\database.schema')
))
elif platform.system() == 'Linux':
app.config.update(dict(
DATABASE=os.path.join(app.root_path, 'database/database.db'),
SCHEMA=os.path.join(app.root_path, 'database/database.schema')
))
else:
print('Unsupported operating system')
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
import server
import database_helpers
from app import server
from app import database_helpers
@app.teardown_appcontext
def close_database_connection(error):
if hasattr(g, 'db'):
g.db.close()
run.py
from app import app
app.run(debug=True)
The project directory tree is:
Aucun commentaire:
Enregistrer un commentaire