diff --git a/cps/kobo.py b/cps/kobo.py index db16fe9a..2acecb6c 100644 --- a/cps/kobo.py +++ b/cps/kobo.py @@ -33,7 +33,7 @@ from sqlalchemy import func from . import config, logger, kobo_auth, db, helper from .web import download_required -kobo = Blueprint("kobo", __name__, url_prefix='/kobo/') +kobo = Blueprint("kobo", __name__, url_prefix="/kobo/") kobo_auth.disable_failed_auth_redirect_for_blueprint(kobo) kobo_auth.register_url_value_preprocessor(kobo) @@ -218,7 +218,12 @@ def HandleMetadataRequest(book_uuid): def get_download_url_for_book(book, book_format): - return url_for("web.download_link", book_id=book.id, book_format=book_format.lower(), _external = True) + return url_for( + "web.download_link", + book_id=book.id, + book_format=book_format.lower(), + _external=True, + ) def create_book_entitlement(book): @@ -350,10 +355,12 @@ def HandleCoverImageRequest(book_uuid, horizontal, vertical, jpeg_quality, monoc return make_response() return book_cover + @kobo.route("") def TopLevelEndpoint(): return make_response(jsonify({})) + @kobo.route("/v1/user/profile") @kobo.route("/v1/user/loyalty/benefits") @kobo.route("/v1/analytics/gettests/", methods=["GET", "POST"]) @@ -388,7 +395,9 @@ def HandleAuthRequest(): @kobo.route("/v1/initialization") def HandleInitRequest(): - resources = NATIVE_KOBO_RESOURCES(calibre_web_url=url_for("web.index", _external=True).strip("/")) + resources = NATIVE_KOBO_RESOURCES( + calibre_web_url=url_for("web.index", _external=True).strip("/") + ) response = make_response(jsonify({"Resources": resources})) response.headers["x-kobo-apitoken"] = "e30=" return response diff --git a/cps/kobo_auth.py b/cps/kobo_auth.py index fef92599..0b9eba6e 100644 --- a/cps/kobo_auth.py +++ b/cps/kobo_auth.py @@ -70,10 +70,11 @@ from .web import render_title_template log = logger.create() + def register_url_value_preprocessor(kobo): @kobo.url_value_preprocessor def pop_auth_token(endpoint, values): - g.auth_token = values.pop('auth_token') + g.auth_token = values.pop("auth_token") def disable_failed_auth_redirect_for_blueprint(bp): @@ -82,31 +83,44 @@ def disable_failed_auth_redirect_for_blueprint(bp): @lm.request_loader def load_user_from_kobo_request(request): - if 'auth_token' in g: - auth_token = g.get('auth_token') - user = ub.session.query(ub.User).join(ub.RemoteAuthToken).filter(ub.RemoteAuthToken.auth_token == auth_token).first() + if "auth_token" in g: + auth_token = g.get("auth_token") + user = ( + ub.session.query(ub.User) + .join(ub.RemoteAuthToken) + .filter(ub.RemoteAuthToken.auth_token == auth_token) + .first() + ) if user is not None: login_user(user) return user log.info("Received Kobo request without a recognizable auth token.") return None -kobo_auth = Blueprint("kobo_auth", __name__, url_prefix='/kobo_auth') -@kobo_auth.route('/generate_auth_token') +kobo_auth = Blueprint("kobo_auth", __name__, url_prefix="/kobo_auth") + + +@kobo_auth.route("/generate_auth_token") @login_required def generate_auth_token(): # Invalidate any prevously generated Kobo Auth token for this user. - ub.session.query(ub.RemoteAuthToken).filter(ub.RemoteAuthToken.user_id == current_user.id).delete() + ub.session.query(ub.RemoteAuthToken).filter( + ub.RemoteAuthToken.user_id == current_user.id + ).delete() auth_token = ub.RemoteAuthToken() auth_token.user_id = current_user.id auth_token.expiration = datetime.max - auth_token.auth_token = (hexlify(urandom(16))).decode('utf-8') + auth_token.auth_token = (hexlify(urandom(16))).decode("utf-8") ub.session.add(auth_token) ub.session.commit() - - return render_title_template('generate_kobo_auth_url.html', title=_(u"Kobo Set-up"), - kobo_auth_url=url_for("kobo.TopLevelEndpoint", auth_token=auth_token.auth_token, _external=True)) \ No newline at end of file + return render_title_template( + "generate_kobo_auth_url.html", + title=_(u"Kobo Set-up"), + kobo_auth_url=url_for( + "kobo.TopLevelEndpoint", auth_token=auth_token.auth_token, _external=True + ), + )