我正在开发一个使用socket.io和flask api的应用程序。
我正在使用Blueprint与我的API一起使用
我已经测试了我的API端点并按预期工作(直到我没有从运行整个项目的app.py文件中导入套接字代码为止),
我测试了我的socket代码,它按预期工作,但现在我发现我的API端点抛出了404错误,更奇怪的是postman stats代码即将达到200,但浏览器呈现为404
我不明白是什么让我的端点在导入套接字代码后消失了(套接字代码工作正常)
我在下面添加我的一些文件,如果需要更多信息,请在评论中告诉我
EDIT1:我发现在我的两张蓝图中,只有Auth blueprint api停止工作,而另一张没有使用前缀的蓝图工作正常
app.py file
import sys
sys.path.insert(0, '.')
from flask import (Flask,
render_template)
from flask_sqlalchemy import SQLAlchemy
from os import getcwd, environ
from os.path import join
from flask_cors import CORS, cross_origin
from flask_socketio import (
SocketIO,
)
app = Flask(__name__)
app.config['DEBUG']=True #--todo--
CORS(app, support_credentials=True)
PATH_TO_CONFIG = join(getcwd(), 'config.py')
app.config.from_pyfile(PATH_TO_CONFIG)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
PG = environ.get("DATABASE_URL")
PATH_TO_LOCAL_DB = join(getcwd(), 'db.sqlite')
if PG is None or PG=="":
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + \
PATH_TO_LOCAL_DB
else:
app.config['SQLALCHEMY_DATABASE_URI'] = PG
db = SQLAlchemy(app)
@app.errorhandler(404)
def error404(error):
return render_template('404page.html')
# Importing BluePrint
from auth import auth_bp
from general import general_bp
# Registring BluePrint
app.register_blueprint(auth_bp, url_prefix='/auth')
app.register_blueprint(general_bp, url_prefix='')
# Heroku Port Configure
PORT = environ.get("PORT")
socketio = SocketIO(app, cors_allowed_origins='*')
from sockets import * ## If comment out this line then API endpoints are working as expected!!
if __name__ == "__main__":
#app.run()
socketio.run(app, host='127.0.0.1', port=PORT)
socket.py
文件,您可能不需要查看此文件(仅供参考)
from app import db, socketio
from dryFunctions import *
from flask_socketio import (
send,
emit,
join_room,
leave_room,
)
# internal
from datetime import timedelta
@socketio.on('join', namespace='/rooms_namespace')
def on_join(data):
"""
verify if attendence is success
then get them join in that room name with otp added
now broadcast to that room
"""
try:
data = json.loads(data) # now it's dict form, before it was string
except:
data = {}
status = data.get('status')
if status!='Success':
return ''
room = '-'.join(data.get('event_name').split(' ')) + '-' + str(data['otp'])
username = data.get("username") #add this to payLoad (clientside)
join_room(room)
mymessage = {
'username':data.get('username'),
'RoomName':room
}
broadcast_choice_ = Events.query.filter_by(otp = data['otp']).first().broadcast_choice
if broadcast_choice_:
emit('join_room', json.dumps(mymessage), room=room)
@socketio.on('admin_stats', namespace='/admin_namespace')
def made_for_admin():
"""
To tell flask-SocketIO that there exsists a namespace named admin_namespace
then we can use it in functions written below. Or else error will be thrown.
"""
#admin_stats is mapped with admin_listen
pass
@socketio.on('attendence_request', namespace='/attendence_namespace')
def take_attendence_from_user(json_):
#attendence_request is mapped with attendence_result at now
"""
This will mark attendence if valid or return failure reason
at the namespace 'attendence_namespace' and this is done privately for
the user.
It also broadcasts the attendence result to
"""
emit('attendence_result', json.dumps(json_)) #json is emited now, privately
otp_ = json_.get('otp')
token_ = json_.get('token')
user_detail = user_info(token_)
if user_detail != 'AuthFail':
user_info_ = user_info(token_)
email_ = user_info_.get("email")
userId = user_info_.get('id')
datetime_ = datetime.datetime.now()
try: # Integer, else error thrown from distance function
# user_latitude = json_['latitude']
# user_longitude = json_['longitude'] # send them from client
user_latitude = json_['latitude']
user_latitude = validFloat(user_latitude)
user_longitude = json_['longitude'] # send them from client
user_longitude = validFloat(user_longitude)
# Not Making Massive Changes, Though some code below can be removed.
assert type(user_latitude)==type(float()) or type(user_latitude)==type(int())
assert type(user_longitude)==type(float()) or type(user_longitude)==type(int())
except:
user_latitude = None
user_longitude = None
try:
event_query = Events.query.filter_by(otp=otp_).first()
admin_latitude = event_query.latitude
admin_longitude = event_query.longitude
location_range_ = event_query.location_range
if location_range_==-1 or admin_latitude==-1.1 or admin_longitude==-1.1: # Lat and Long failed to come or Distance Range not given
distance_ = 10
location_range_ = 0
else:
if user_latitude==None or user_longitude==None:
location_range_=-1
distance_ = 10
location_range_ = 0
else:
distance_ = distance([user_latitude, user_longitude], [admin_latitude, admin_longitude])
if distance_>location_range_:
status_ = 0
else:
status_ = 1
admin_datetime = event_query.creation_date
ending_time_delta_ = event_query.ending_time_delta
datetime_check = admin_datetime + timedelta(minutes=ending_time_delta_)
communityId = event_query.communityId
if datetime_ > datetime_check: # After Event has Ended
time_constraint = 1
else:
time_constraint = 0
event_id_ = event_query.eventId #Earlier this was in catch block, now more otp_ associated things added in this catch
except Exception as e: # Event Doesn't Exsist
event_id_ = -1
allow_forward = 0 # Flag to allow only Successfull Attendances to be passed to Admin
if event_id_>=0:
if time_constraint==1: # Keep Failures Before then the Success.
Reason='Event has Ended now'
StatusCode=423
payLoad = {
'Status':'Fail',
'Reason':Reason,
'StatusCode':StatusCode
}
elif status_==0: # We are taking attendance but with status 0, which means absent \
# this is because location failure can occur. So we can consider them present.
Reason='You are far away from Event Location'
StatusCode=400
payLoad = {
'Status':'Fail',
'Reason':Reason,
'StatusCode':StatusCode
}
new_attendence = Attendances(userId=userId, eventId=event_id_, \
communityId=communityId,
datetime=datetime_, status=status_)
db.session.add(new_attendence)
db.session.commit()
elif status_==1:
already_given = False
# Checking if user has already given attendence
event_attendences = Attendances.query.filter_by(eventId=event_id_).all()
for user in event_attendences:
if user.userId == userId and user.status in ['t', 'True', True, '1', 1]:
Reason='Attendence Already Given and Present'
StatusCode=400
payLoad = {
'Status':'Success',
'Reason':Reason,
'StatusCode':StatusCode
}
already_given = True
break
if not already_given:
Reason='Attendence updated'
StatusCode=200
new_attendence = Attendances(userId=userId, eventId=event_id_, \
communityId=communityId,
datetime=datetime_, status=status_)
db.session.add(new_attendence)
db.session.commit()
payLoad = {
'Status':'Success',
'Reason':Reason,
'StatusCode':StatusCode
}
allow_forward = 1
else:
Reason='Server Error'
StatusCode=500
payLoad = {
'Status':'Fail',
'Reason': Reason,
'StatusCode':StatusCode
}
else:
print(otp_, )
Reason='No such Event'
StatusCode=404
payLoad = {
'Status':'Fail',
'Reason':Reason,
'StatusCode':StatusCode
}
else:
Reason='AuthFail'
StatusCode=401
payLoad = {
'Status':'Fail',
'Reason': Reason,
'StatusCode':StatusCode
}
emit('attendence_result', json.dumps(payLoad)) #json is emited now, privately
try:
# Now Emitting to Admin Namespace and from there it's \
# sended to specific room also if admin has allowed this.
# Checking if this user has given attendance for this event (otp) and don't forward it to admin in case attandance is given
first_time = True
first_time_count = 0 # Because Attendence is already given above, so 1 entry is already there
try:
user_attendences = Attendances.query.filter_by(email=email_)
for EVENT in user_attendences:
if EVENT.event_otp==otp_:
first_time_count+=1
if first_time_count==2:
first_time = False
break
except: # Will be handled below
pass
if first_time == True:
if status_:
# negative otp means ignore it, wrong entry
# here recheck for otp is not needed, so can be removed
try:
if allow_forward==0:
raise ValueError('Attendance Failure')
event = Events.query.filter_by(otp=otp_).first()
event_name_ = event.event_name
admin_email_ = event.admin_email
attendence_status = 'Success' # This param is for admin only
except:
event_name_ = ''
admin_email_ = ''
otp_ = -1
attendence_status = 'Failure' # This param is for admin only
payLoad = {
'email':user_info(token_).get("email"),
'username':user_info(token_).get("username"),
'datetime':str(datetime_),
'status': attendence_status,
'event_name':event_name_,
'admin_email':admin_email_, # use this to send to specific admin
'otp':otp_,
'Reason':Reason,
'StatusCode':StatusCode
}
emit('admin_listen', json.dumps(payLoad), namespace='/admin_namespace')
else:
raise ValueError('Status failure')
except:
payLoad = {
'email':'',
'username':'',
'datetime':str(datetime.datetime.now()),
'status': 'Failure',
'event_name':'',
'admin_email':'',
'otp':-1,
'Reason':Reason,
'StatusCode':StatusCode
}
emit('admin_listen', json.dumps(payLoad), namespace='/admin_namespace')
目前没有回答
相关问题 更多 >
编程相关推荐