问题描述
我制作了一个 api 文件并运行它，但错误不断弹出。用的是jsonify，没有dump，Python版本是最新的。
api 文件：
from flask import request,current_app,jsonify,make_response
from flask_restx import Resource,Api,Namespace,fields
from flask_jwt_extended import *
from sqlalchemy import text
MyPage = Namespace(
name='Mypage',description="마이페이지 정보를 관리하는 API 모음."
)
@MyPage.route('/user_info')
class User_Info(Resource):
@jwt_required
def post(self):
current_user = get_jwt_identity()
user_id = request.form['id']
if current_user is None:
return make_response(jsonify({
"result": "fail","contents": "잘못된 토큰값입니다."
}),401)
else:
user = current_app.database.execute(text("""
SELECT * FROM User WHERE id= :user_id ;
"""),{
'user_id': user_id
}).fetchone()
if user is None:
return [],204
else:
d_user = dict(user)
if d_user['is_author'] == 0:
d_user['is_author'] = False
else:
d_user['is_author'] = True
return jsonify({
'id': d_user['id'],'name': d_user['name'],'profile': d_user['profile'],'is_author': d_user['is_author'],'email': d_user['email'],'introduction': d_user['introduction'],'coin': d_user['coin']
})
@MyPage.route('/my_writing')
class Write_By_Me(Resource):
@jwt_required
def post(self):
current_user = get_jwt_identity()
user_id = request.form['id']
if current_user is None:
return make_response(jsonify({
"result": "fail",401)
else:
writing_list = current_app.database.execute(text("""
SELECT * FROM Writing Where author_id = :author_id ORDER BY id DESC;
"""),{
'author_id': user_id
}).fetchall()
series_info = current_app.database.execute(text("""
SELECT title,is_end FROM Series WHERE id = :series_id;
"""),{
'series_id': d_writing['series_id']
}).fetchone()
d_writing['series_name'] = series_info['title']
if series_info['is_end'] == 1:
d_writing['is_end'] = True
else:
d_writing['is_end'] = False
d_writing_list.append(d_writing)
return jsonify(
[{
'id': writing['id'],'series_id': writing['series_id'],'series_name': writing['series_name'],'episode_num': writing['episode_num'],'title': writing['title'],'image': writing['image'],'date': writing['date'],'hits': writing['hits'],'comment_num': writing['comment_num'],'permission': writing['permission'],'is_end': writing['is_end'],'author_id': writing['author_id'],'contents': writing['contents']
} for writing in d_writing_list]
)
else:
return jsonify([])
@MyPage.route('/subscribe_author')
class Subscribe_List_Author(Resource):
@jwt_required
def post(self):
current_user = get_jwt_identity()
user_id = request.form['id']
if current_user is None:
# 3. 401(Unauthorized)를 리턴한다.
return make_response(jsonify({
"result": "fail",401)
else:
subscribe_list = current_app.database.execute(text("""
SELECT * FROM Subscribe_Author
WHERE user_id = :user_id ;
"""),{
'user_id': user_id
}).fetchall()
if len(subscribe_list) > 0:
author_query_string = ''
for subscribe in subscribe_list:
author_query_string += 'id = '
author_query_string += str(subscribe['author_id'])
author_query_string += ' OR '
author_query_string = author_query_string[0:len(author_query_string) - 4]
author_query_string = 'SELECT * FROM User WHERE ' + author_query_string + ";"
author_list = current_app.database.execute(text(author_query_string)).fetchall()
return jsonify(
[{
'id': author['id'],'name': author['name'],'profile': author['profile'],'series_num': author['series_num'],'recent_update': author['recent_update'],'zzimkkong': author['zzimkkong']
} for author in author_list]
)
else:
return jsonify([])
@MyPage.route('/add_subscribe_author')
class Add_Subscribe_Author(Resource):
@jwt_required
def post(self):
current_user = get_jwt_identity()
# user_id,author_id를 가져옴.
user_id = request.form['user_id']
author_id = request.form['author_id']
is_subscribe = request.form['is_subscribe']
if current_user is None:
# 401(Unauthorized)를 리턴한다.
return make_response(jsonify({
"result": "fail",401)
# 토큰값이 정상인 경우
else:
if is_subscribe == 'T':
duplication_value = current_app.database.execute(text("""
SELECT * FROM Subscribe_Author WHERE user_id=:user_id AND author_id=:author_id;
"""),{
'user_id': user_id,'author_id': author_id
}).fetchall()
if len(duplication_value) == 0:
current_app.database.execute(text("""
INSERT INTO Subscribe_Author (user_id,author_id) VALUES (:user_id,:author_id);
"""),{
'user_id': user_id,'author_id': author_id
})
current_app.database.execute(text("""
UPDATE User SET zzimkkong=zzimkkong+1 WHERE id = :author_id ;
"""),{
'author_id': author_id
})
return jsonify(
{
'result': 'success','contents': 'success'
}
)
else:
return make_response(jsonify(
{
"result": "fail","contents": "중복된 값이 있습니다."
}
),202)
else:
duplication_value = current_app.database.execute(text("""
SELECT * FROM Subscribe_Author WHERE user_id=:user_id AND author_id=:author_id;
"""),'author_id': author_id
}).fetchall()
if len(duplication_value) > 0:
current_app.database.execute(text("""
DELETE FROM Subscribe_Author WHERE user_id = :user_id AND author_id = :author_id;
"""),'author_id': author_id
})
current_app.database.execute(text("""
UPDATE User SET zzimkkong=zzimkkong-1 WHERE id = :author_id ;
"""),'contents': 'success'
}
)
else:
return make_response(jsonify(
{
"result": "fail","contents": "해당하는 값이 없습니다."
}
),202)
@MyPage.route('/subscribe_series')
class Subscribe_List_Series(Resource):
@jwt_required
def post(self):
current_user = get_jwt_identity()
user_id = request.form['id']
if current_user is None:
# 401(Unauthorized)를 리턴한다.
return make_response(jsonify({
"result": "fail",401)
else:
subscribe_list = current_app.database.execute(text("""
SELECT * FROM Subscribe_Series
WHERE user_id = :user_id ;
"""),{
'user_id': user_id
}).fetchall()
if len(subscribe_list) > 0:
series_query_string = ''
for subscribe in subscribe_list:
series_query_string += 'id = '
series_query_string += str(subscribe['series_id'])
series_query_string += ' OR '
series_query_string = series_query_string[0:len(series_query_string) - 4]
series_query_string = 'SELECT * FROM Series WHERE ' + series_query_string + ";"
series_list = current_app.database.execute(text(series_query_string)).fetchall()
d_series_list = []
for series in series_list:
d_series = dict(series)
if d_series['is_end'] == 1:
d_series['is_end'] = True
else:
d_series['is_end'] = False
d_series_list.append(d_series)
return jsonify(
[{
'id': series['id'],'author_id': series['author_id'],'author_name': series['author_name'],'title': series['title'],'image': series['image'],'date': series['date'],'introduction': series['introduction'],'hash_tag': series['hash_tag'],'hits': series['hits'],'comment_num': series['comment_num'],'recent_update': series['recent_update'],'zzimkkong': series['zzimkkong'],'is_end': series['is_end'],'writing_num': series['writing_num']
} for series in d_series_list]
)
else:
return jsonify([])
@MyPage.route('/add_subscribe_series')
class Add_Subscribe_Series(Resource):
@jwt_required
def post(self):
"""현재 유저의 시리즈 구독 목록에 새로운 시리즈를 추가함."""
# 토큰값을 가져옴.
current_user = get_jwt_identity()
# user_id,series_id를 가져옴.
user_id = request.form['user_id']
series_id = request.form['series_id']
is_subscribe = request.form['is_subscribe']
# 토큰값이 잘못된 경우
if current_user is None:
# 401(Unauthorized)를 리턴한다.
return make_response(jsonify({
"result": "fail",401)
# 토큰값이 정상인 경우
else:
# 구독 처리를 하는 경우
if is_subscribe == 'T':
# 1. 중복된 값을 찾아본다.
duplication_value = current_app.database.execute(text("""
SELECT * FROM Subscribe_Series WHERE user_id=:user_id AND series_id=:series_id;
"""),'series_id': series_id
}).fetchall()
# 2. 중복된 값이 없다면 등록을 진행하고 success를 리턴한다.
if len(duplication_value) == 0:
current_app.database.execute(text("""
INSERT INTO Subscribe_Series (user_id,series_id) VALUES (:user_id,:series_id);
"""),'series_id': series_id
})
# 추가 : 시리즈 DB의 zzimkkong 값을 하나 증가시킨다.
current_app.database.execute(text("""
UPDATE Series SET zzimkkong=zzimkkong+1 WHERE id = :series_id ;
"""),{
'series_id': series_id
})
return jsonify(
{
'result': 'success','contents': 'success'
}
)
# 2. 중복된 값이 있다면 오류를 리턴한다.
else:
return make_response(jsonify(
{
"result": "fail",202)
# 구독 취소를 하는 경우
else:
# 1. 해당 값이 있는지 찾아본다.
duplication_value = current_app.database.execute(text("""
SELECT * FROM Subscribe_Series WHERE user_id=:user_id AND series_id=:series_id;
"""),'series_id': series_id
}).fetchall()
# 2. 해당 값이 있다면
if len(duplication_value) > 0:
# 2. 구독 취소 처리를 하고 success 출력.
current_app.database.execute(text("""
DELETE FROM Subscribe_Series WHERE user_id = :user_id AND series_id = :series_id;
"""),'series_id': series_id
})
# 추가 : 시리즈 DB의 zzimkkong 값을 하나 감소시킨다.
current_app.database.execute(text("""
UPDATE Series SET zzimkkong=zzimkkong-1 WHERE id = :series_id ;
"""),'contents': 'success'
}
)
# 해당 값이 없다면
else:
return make_response(jsonify(
result="fail",contents="해당하는 값이 없습니다."
),202)
@MyPage.route('/update_introduction')
class User_Intro(Resource):
@jwt_required
def post(self):
"""현재 유저의 정보를 가져옴."""
# 0. 토큰값을 가져옴.
current_user = get_jwt_identity()
# 1. 필요한 값들을 parameter로 받아옴.
user_id = request.form['id']
intro = request.form['introduction']
user_name = request.form['name']
# 2. 토큰값이 잘못된 경우
if current_user is None:
# 3. 401(Unauthorized)를 리턴한다.
return make_response(jsonify({
"result": "fail",401)
# 2. 토큰값이 정상인 경우
else:
# 3. 유저 정보를 가져옴.
current_app.database.execute(text("""
UPDATE User SET introduction = :intro,name = :name WHERE id = :user_id ;
"""),{
'user_id': user_id,'intro': intro,'name': user_name
})
return jsonify(
{
'result': 'success','contents': 'success'
}
)
@MyPage.route('/my_series')
class Series_By_Me(Resource):
@jwt_required
def post(self):
"""내가 쓴 시리즈 정보를 불러옴."""
# 0. 토큰값을 가져옴.
current_user = get_jwt_identity()
# 1. 필요한 값들을 parameter로 받아옴.
user_id = request.form['id']
# 2. 토큰값이 잘못된 경우
if current_user is None:
# 3. 401(Unauthorized)를 리턴한다.
return jsonify({
"result": "fail",401
# 2. 토큰값이 정상인 경우
else:
# 3. 최신순 정렬인 경우 -> id 내림차순 순서대로 쿼리.
writing_list = current_app.database.execute(text("""
SELECT * FROM Series Where author_id = :author_id ORDER BY id DESC;
"""),{
'author_id': user_id
}).fetchall()
d_series_list = []
for series in writing_list:
d_series = dict(series)
if d_series['is_end'] == 1:
d_series['is_end'] = True
else:
d_series['is_end'] = False
d_series_list.append(d_series)
if len(writing_list) > 0:
return jsonify(
[{
'id': series['id'],'writing_num': series['writing_num']
} for series in writing_list]
)
else:
return jsonify([])
@MyPage.route('/finish_series')
class Finish_Series(Resource):
@jwt_required
def post(self):
"""시리즈를 완결시킴."""
# 토큰값을 가져옴.
current_user = get_jwt_identity()
# user_id,series_id를 가져옴.
user_id = request.form['id']
series_id = request.form['series_id']
is_ended = request.form['is_end']
# 토큰값이 잘못된 경우
if current_user is None:
# 401(Unauthorized)를 리턴한다.
return jsonify({
"result": "fail",401
# 토큰값이 정상인 경우
else:
# 완결인 경우
if is_ended == 'T':
current_app.database.execute(text("""
UPDATE Series SET is_end = 1 WHERE id = :series_id ;
"""),{
'series_id': series_id
})
elif is_ended == 'F':
current_app.database.execute(text("""
UPDATE Series SET is_end = 0 WHERE id = :series_id ;
"""),{
'series_id': series_id
})
return jsonify(
{
'result': 'success','contents': 'success'
}
)
在 postman 上运行时出错：
Traceback (most recent call last):
File "/usr/local/lib/python3.9/site-packages/flask/app.py",line 2464,in __call__
return self.wsgi_app(environ,start_response)
File "/usr/local/lib/python3.9/site-packages/flask/app.py",line 2450,in wsgi_app
response = self.handle_exception(e)
File "/usr/local/lib/python3.9/site-packages/flask_cors/extension.py",line 165,in wrapped_function
return cors_after_request(app.make_response(f(*args,**kwargs)))
File "/usr/local/lib/python3.9/site-packages/flask_restx/api.py",line 638,in error_router
return original_handler(f)
File "/usr/local/lib/python3.9/site-packages/flask/app.py",line 1867,in handle_exception
reraise(exc_type,exc_value,tb)
File "/usr/local/lib/python3.9/site-packages/flask/_compat.py",line 39,in reraise
raise value
File "/usr/local/lib/python3.9/site-packages/flask_restx/api.py",line 636,in error_router
return self.handle_error(e)
File "/usr/local/lib/python3.9/site-packages/flask/app.py",line 2447,in wsgi_app
response = self.full_dispatch_request()
File "/usr/local/lib/python3.9/site-packages/flask/app.py",line 1952,in full_dispatch_request
rv = self.handle_user_exception(e)
File "/usr/local/lib/python3.9/site-packages/flask_cors/extension.py",line 1821,in handle_user_exception
reraise(exc_type,line 1950,in full_dispatch_request
rv = self.dispatch_request()
File "/usr/local/lib/python3.9/site-packages/flask/app.py",line 1936,in dispatch_request
return self.view_functions[rule.endpoint](**req.view_args)
File "/usr/local/lib/python3.9/site-packages/flask_restx/api.py",line 379,in wrapper
return self.make_response(data,code,headers=headers)
File "/usr/local/lib/python3.9/site-packages/flask_restx/api.py",line 402,in make_response
resp = self.representations[mediatype](data,*args,**kwargs)
File "/usr/local/lib/python3.9/site-packages/flask_restx/representations.py",line 25,in output_json
dumped = dumps(data,**settings) + "\n"
File "/usr/local/lib/python3.9/json/__init__.py",line 234,in dumps
return cls(
File "/usr/local/lib/python3.9/json/encoder.py",line 201,in encode
chunks = list(chunks)
File "/usr/local/lib/python3.9/json/encoder.py",line 438,in _iterencode
o = _default(o)
File "/usr/local/lib/python3.9/json/encoder.py",line 179,in default
raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type function is not JSON serializable
--> 任何文件中都没有 json.dump。
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）