I keep getting AttributeError: 'str' object has no attribute 'request' with google APIs in flask
I keep getting AttributeError: 'str' object has no attribute 'request' with google APIs in flask
我正在尝试通过 Flask 与 android-management-api 进行交互。
每次我 运行 遇到一个我不理解的错误，因为我对编码很陌生
调用device_list = androidmanagement.enterprises().devices().list(parent=enterprise_name, pageSize=200).execute() 时出现错误
我只是不明白为什么我会收到这个错误。
如果有人能解释这是怎么发生的，我会很高兴。
非常感谢
我的代码在app.py
import json
import os
import sqlite3
from datetime import datetime
# flask imports
from flask import Flask, redirect, url_for, render_template, request
from flask_login import (
LoginManager,
current_user,
login_user,
logout_user,
)
# Oauth imports
from oauthlib.oauth2 import WebApplicationClient
import requests
# database imports
from db import init_db_command
from user import User
# google imports
from googleapiclient.discovery import build
# '''End of imports'''
# '''Start of settings'''
# Configuration load environment settings
GOOGLE_CLIENT_ID = os.environ.get("GOOGLE_CLIENT_ID", None)
GOOGLE_CLIENT_SECRET = os.environ.get("GOOGLE_CLIENT_SECRET", None)
GOOGLE_DISCOVERY_URL = (
"https://accounts.google.com/.well-known/openid-configuration"
)
# ''' Start of local variables'''
cloud_project_id = 'hidden'
enterprise_name = 'hidden'
# ''' End of Local variables'''
# ''' Start build service object '''
API_KEY = os.environ.get(GOOGLE_API_KEY)
API_SERVICE_NAME = 'androidmanagement'
API_VERSION = 'v1'
androidmanagement = build(API_SERVICE_NAME, API_VERSION, API_KEY)
# ''' End of service object '''
# Flask app setup
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY") or os.urandom(24)
# User session management setup
# https://flask-login.readthedocs.io/en/latest
login_manager = LoginManager()
login_manager.init_app(app)
# Naive database setup
try:
init_db_command()
except sqlite3.OperationalError:
# Assume it's already been created
pass
# OAuth 2 client setup
client = WebApplicationClient(GOOGLE_CLIENT_ID)
def get_google_provider_cfg():
return requests.get(GOOGLE_DISCOVERY_URL).json()
# Flask-Login helper to retrieve a user from our db
@login_manager.user_loader
def load_user(user_id):
return User.get(user_id)
# '''End of settings'''
# '''All view routing'''
# index
@app.route('/')
def index():
if current_user.is_authenticated:
return redirect(url_for('home'))
else:
return redirect(url_for('loginuser'))
# """Starts the login flow."""
@app.route("/login")
def login():
# Find out what URL to hit for Google login
google_provider_cfg = get_google_provider_cfg()
authorization_endpoint = google_provider_cfg["authorization_endpoint"]
# Use library to construct the request for Google login and provide
# scopes that let you retrieve user's profile from Google
request_uri = client.prepare_request_uri(
authorization_endpoint,
redirect_uri=request.base_url + "/callback",
scope=["openid", "email", "profile"],
)
return redirect(request_uri)
# auth flow callback.
@app.route("/login/callback")
def callback():
# Get authorization code Google sent back to you
code = request.args.get("code")
# Find out what URL to hit to get tokens that allow you to ask for
# things on behalf of a user
google_provider_cfg = get_google_provider_cfg()
token_endpoint = google_provider_cfg["token_endpoint"]
# Prepare and send request to get tokens! Yay tokens!
token_url, headers, body = client.prepare_token_request(
token_endpoint,
authorization_response=request.url,
redirect_url=request.base_url,
code=code,
)
token_response = requests.post(
token_url,
headers=headers,
data=body,
auth=(GOOGLE_CLIENT_ID, GOOGLE_CLIENT_SECRET),
)
# Parse the tokens!
client.parse_request_body_response(json.dumps(token_response.json()))
# Now that we have tokens (yay) let's find and hit URL
# from Google that gives you user's profile information,
# including their Google Profile Image and Email
userinfo_endpoint = google_provider_cfg["userinfo_endpoint"]
uri, headers, body = client.add_token(userinfo_endpoint)
userinfo_response = requests.get(uri, headers=headers, data=body)
# We want to make sure their email is verified.
# The user authenticated with Google, authorized our
# app, and now we've verified their email through Google!
if userinfo_response.json().get("email_verified"):
unique_id = userinfo_response.json()["sub"]
users_email = userinfo_response.json()["email"]
picture = userinfo_response.json()["picture"]
users_name = userinfo_response.json()["given_name"]
else:
return "User email not available or not verified by Google.", 400
# Create a user in our db with the information provided
# by Google
user = User(
id_=unique_id, name=users_name, email=users_email, profile_pic=picture
)
# Doesn't exist? Add to database
if not User.get(unique_id):
User.create(unique_id, users_name, users_email, picture)
# Begin user session by logging the user in
login_user(user)
# Send user back to homepage
return redirect(url_for("home"))
@app.route('/logout')
def logout():
if current_user.is_authenticated:
logout_user()
return render_template(
'logoutuser.html',
title='log out',
year=datetime.now().year,
)
else:
return redirect(url_for("index"))
# Renders the login page.
@app.route('/loginuser')
def loginuser():
if current_user.is_authenticated:
return redirect(url_for('home'))
else:
return render_template('loginuser.html')
# base web routing start here
# Renders the home page.
# @app.route('/')
@app.route('/home')
def home():
if current_user.is_authenticated:
return render_template(
'index.html',
title='Home Page',
year=datetime.now().year,
)
else:
return redirect(url_for('loginuser'))
# Renders the devices page.
@app.route('/devices')
def devices():
if current_user.is_authenticated:
device_list = androidmanagement.enterprises().devices().list(parent=enterprise_name, pageSize=200).execute()
return render_template(
'devices.html',
title='Devices',
year=datetime.now().year,
device_list=device_list.items()
)
else:
return redirect(url_for('loginuser'))
# Renders the policies page.
@app.route('/policies')
def policies():
if current_user.is_authenticated:
return render_template(
'policies.html',
title='Policies',
year=datetime.now().year,
)
else:
return redirect(url_for('loginuser'))
# Renders the enterprise page.
@app.route('/enterprises')
def enterprises():
if current_user.is_authenticated:
return render_template(
'enterprises.html',
title='Enterprises',
year=datetime.now().year,
)
else:
return redirect(url_for('loginuser'))
# Renders the contact page.
@app.route('/contact')
def contact():
if current_user.is_authenticated:
return render_template(
'contact.html',
title='Contact',
year=datetime.now().year,
message='ICS-Vertex'
)
else:
return redirect(url_for('loginuser'))
# Renders the about page.
@app.route('/about')
def about():
if current_user.is_authenticated:
return render_template(
'about.html',
title='About',
year=datetime.now().year,
message='Your application description page.'
)
else:
return redirect(url_for('loginuser'))
# '''end of view routing'''
# Starts application with all above code
if __name__ == '__main__':
# app.run()
app.debug = True
app.run(ssl_context="adhoc", debug=True)
我的堆栈跟踪说明如下：
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 2069, in __call__
def __call__(self, environ: dict, start_response: t.Callable) -> t.Any:
"""The WSGI server calls the Flask application object as the
WSGI application. This calls :meth:`wsgi_app`, which can be
wrapped to apply middleware.
"""
return self.wsgi_app(environ, start_response)
def _request_blueprints(self) -> t.Iterable[str]:
if _request_ctx_stack.top.request.blueprint is None:
return []
else:
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 2054, in wsgi_app
try:Open an interactive python shell in this frame
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 2051, in wsgi_app
ctx = self.request_context(environ)
error: t.Optional[BaseException] = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 1501, in full_dispatch_request
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
def finalize_request(
self,
rv: t.Union[ResponseReturnValue, HTTPException],
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 1499, in full_dispatch_request
self.try_trigger_before_first_request_functions()
try:
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
def finalize_request(
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 1485, in dispatch_request
getattr(rule, "provide_automatic_options", False)
and req.method == "OPTIONS"
):
return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint
return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)
def full_dispatch_request(self) -> Response:
"""Dispatches the request and on top of that performs request"""
pre and postprocessing as well as HTTP exception catching and
error handling.
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\app.py", line 226, in devices
# Renders the devices page.
@app.route('/devices')
def devices():
if current_user.is_authenticated:
device_list = androidmanagement.enterprises().devices().list(parent=enterprise_name, pageSize=200).execute()
return render_template(
'devices.html',
title='Devices',
year=datetime.now().year,
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\googleapiclient\_helpers.py", line 134, in positional_wrapper
)
if positional_parameters_enforcement == POSITIONAL_EXCEPTION:
raise TypeError(message)
elif positional_parameters_enforcement == POSITIONAL_WARNING:
logger.warning(message)
return wrapped(*args, **kwargs)
return positional_wrapper
if isinstance(max_positional_args, six.integer_types):
return positional_decorator
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\googleapiclient\http.py", line 920, in execute
)
self.body = parsed.query
self.headers["content-length"] = str(len(self.body))
# Handle retries for server-side errors.
resp, content = _retry_request(
http,
num_retries,
"request",
self._sleep,
self._rand,
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\googleapiclient\http.py", line 191, in _retry_request
)
sleep(sleep_time)
try:
exception = None
resp, content = http.request(uri, method, *args, **kwargs)
# Retry on SSL errors and socket timeout errors.
except _ssl_SSLError as ssl_error:
exception = ssl_error
except socket.timeout as socket_timeout:
# Needs to be before socket.error as it's a subclass of OSError
AttributeError: 'str' object has no attribute 'request'
所以我发现问题是如何出现的：
有 3 个模块在执行相同的任务
google API 客户端
烧瓶要求
请求
这导致代码冲突。
将在我的 API 通话正常后更新
我正在尝试通过 Flask 与 android-management-api 进行交互。 每次我 运行 遇到一个我不理解的错误，因为我对编码很陌生
调用device_list = androidmanagement.enterprises().devices().list(parent=enterprise_name, pageSize=200).execute() 时出现错误
我只是不明白为什么我会收到这个错误。
如果有人能解释这是怎么发生的，我会很高兴。
非常感谢
我的代码在app.py
import json
import os
import sqlite3
from datetime import datetime
# flask imports
from flask import Flask, redirect, url_for, render_template, request
from flask_login import (
LoginManager,
current_user,
login_user,
logout_user,
)
# Oauth imports
from oauthlib.oauth2 import WebApplicationClient
import requests
# database imports
from db import init_db_command
from user import User
# google imports
from googleapiclient.discovery import build
# '''End of imports'''
# '''Start of settings'''
# Configuration load environment settings
GOOGLE_CLIENT_ID = os.environ.get("GOOGLE_CLIENT_ID", None)
GOOGLE_CLIENT_SECRET = os.environ.get("GOOGLE_CLIENT_SECRET", None)
GOOGLE_DISCOVERY_URL = (
"https://accounts.google.com/.well-known/openid-configuration"
)
# ''' Start of local variables'''
cloud_project_id = 'hidden'
enterprise_name = 'hidden'
# ''' End of Local variables'''
# ''' Start build service object '''
API_KEY = os.environ.get(GOOGLE_API_KEY)
API_SERVICE_NAME = 'androidmanagement'
API_VERSION = 'v1'
androidmanagement = build(API_SERVICE_NAME, API_VERSION, API_KEY)
# ''' End of service object '''
# Flask app setup
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY") or os.urandom(24)
# User session management setup
# https://flask-login.readthedocs.io/en/latest
login_manager = LoginManager()
login_manager.init_app(app)
# Naive database setup
try:
init_db_command()
except sqlite3.OperationalError:
# Assume it's already been created
pass
# OAuth 2 client setup
client = WebApplicationClient(GOOGLE_CLIENT_ID)
def get_google_provider_cfg():
return requests.get(GOOGLE_DISCOVERY_URL).json()
# Flask-Login helper to retrieve a user from our db
@login_manager.user_loader
def load_user(user_id):
return User.get(user_id)
# '''End of settings'''
# '''All view routing'''
# index
@app.route('/')
def index():
if current_user.is_authenticated:
return redirect(url_for('home'))
else:
return redirect(url_for('loginuser'))
# """Starts the login flow."""
@app.route("/login")
def login():
# Find out what URL to hit for Google login
google_provider_cfg = get_google_provider_cfg()
authorization_endpoint = google_provider_cfg["authorization_endpoint"]
# Use library to construct the request for Google login and provide
# scopes that let you retrieve user's profile from Google
request_uri = client.prepare_request_uri(
authorization_endpoint,
redirect_uri=request.base_url + "/callback",
scope=["openid", "email", "profile"],
)
return redirect(request_uri)
# auth flow callback.
@app.route("/login/callback")
def callback():
# Get authorization code Google sent back to you
code = request.args.get("code")
# Find out what URL to hit to get tokens that allow you to ask for
# things on behalf of a user
google_provider_cfg = get_google_provider_cfg()
token_endpoint = google_provider_cfg["token_endpoint"]
# Prepare and send request to get tokens! Yay tokens!
token_url, headers, body = client.prepare_token_request(
token_endpoint,
authorization_response=request.url,
redirect_url=request.base_url,
code=code,
)
token_response = requests.post(
token_url,
headers=headers,
data=body,
auth=(GOOGLE_CLIENT_ID, GOOGLE_CLIENT_SECRET),
)
# Parse the tokens!
client.parse_request_body_response(json.dumps(token_response.json()))
# Now that we have tokens (yay) let's find and hit URL
# from Google that gives you user's profile information,
# including their Google Profile Image and Email
userinfo_endpoint = google_provider_cfg["userinfo_endpoint"]
uri, headers, body = client.add_token(userinfo_endpoint)
userinfo_response = requests.get(uri, headers=headers, data=body)
# We want to make sure their email is verified.
# The user authenticated with Google, authorized our
# app, and now we've verified their email through Google!
if userinfo_response.json().get("email_verified"):
unique_id = userinfo_response.json()["sub"]
users_email = userinfo_response.json()["email"]
picture = userinfo_response.json()["picture"]
users_name = userinfo_response.json()["given_name"]
else:
return "User email not available or not verified by Google.", 400
# Create a user in our db with the information provided
# by Google
user = User(
id_=unique_id, name=users_name, email=users_email, profile_pic=picture
)
# Doesn't exist? Add to database
if not User.get(unique_id):
User.create(unique_id, users_name, users_email, picture)
# Begin user session by logging the user in
login_user(user)
# Send user back to homepage
return redirect(url_for("home"))
@app.route('/logout')
def logout():
if current_user.is_authenticated:
logout_user()
return render_template(
'logoutuser.html',
title='log out',
year=datetime.now().year,
)
else:
return redirect(url_for("index"))
# Renders the login page.
@app.route('/loginuser')
def loginuser():
if current_user.is_authenticated:
return redirect(url_for('home'))
else:
return render_template('loginuser.html')
# base web routing start here
# Renders the home page.
# @app.route('/')
@app.route('/home')
def home():
if current_user.is_authenticated:
return render_template(
'index.html',
title='Home Page',
year=datetime.now().year,
)
else:
return redirect(url_for('loginuser'))
# Renders the devices page.
@app.route('/devices')
def devices():
if current_user.is_authenticated:
device_list = androidmanagement.enterprises().devices().list(parent=enterprise_name, pageSize=200).execute()
return render_template(
'devices.html',
title='Devices',
year=datetime.now().year,
device_list=device_list.items()
)
else:
return redirect(url_for('loginuser'))
# Renders the policies page.
@app.route('/policies')
def policies():
if current_user.is_authenticated:
return render_template(
'policies.html',
title='Policies',
year=datetime.now().year,
)
else:
return redirect(url_for('loginuser'))
# Renders the enterprise page.
@app.route('/enterprises')
def enterprises():
if current_user.is_authenticated:
return render_template(
'enterprises.html',
title='Enterprises',
year=datetime.now().year,
)
else:
return redirect(url_for('loginuser'))
# Renders the contact page.
@app.route('/contact')
def contact():
if current_user.is_authenticated:
return render_template(
'contact.html',
title='Contact',
year=datetime.now().year,
message='ICS-Vertex'
)
else:
return redirect(url_for('loginuser'))
# Renders the about page.
@app.route('/about')
def about():
if current_user.is_authenticated:
return render_template(
'about.html',
title='About',
year=datetime.now().year,
message='Your application description page.'
)
else:
return redirect(url_for('loginuser'))
# '''end of view routing'''
# Starts application with all above code
if __name__ == '__main__':
# app.run()
app.debug = True
app.run(ssl_context="adhoc", debug=True)
我的堆栈跟踪说明如下：
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 2069, in __call__
def __call__(self, environ: dict, start_response: t.Callable) -> t.Any:
"""The WSGI server calls the Flask application object as the
WSGI application. This calls :meth:`wsgi_app`, which can be
wrapped to apply middleware.
"""
return self.wsgi_app(environ, start_response)
def _request_blueprints(self) -> t.Iterable[str]:
if _request_ctx_stack.top.request.blueprint is None:
return []
else:
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 2054, in wsgi_app
try:Open an interactive python shell in this frame
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 2051, in wsgi_app
ctx = self.request_context(environ)
error: t.Optional[BaseException] = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 1501, in full_dispatch_request
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
def finalize_request(
self,
rv: t.Union[ResponseReturnValue, HTTPException],
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 1499, in full_dispatch_request
self.try_trigger_before_first_request_functions()
try:
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
def finalize_request(
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\flask\app.py", line 1485, in dispatch_request
getattr(rule, "provide_automatic_options", False)
and req.method == "OPTIONS"
):
return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint
return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)
def full_dispatch_request(self) -> Response:
"""Dispatches the request and on top of that performs request"""
pre and postprocessing as well as HTTP exception catching and
error handling.
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\app.py", line 226, in devices
# Renders the devices page.
@app.route('/devices')
def devices():
if current_user.is_authenticated:
device_list = androidmanagement.enterprises().devices().list(parent=enterprise_name, pageSize=200).execute()
return render_template(
'devices.html',
title='Devices',
year=datetime.now().year,
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\googleapiclient\_helpers.py", line 134, in positional_wrapper
)
if positional_parameters_enforcement == POSITIONAL_EXCEPTION:
raise TypeError(message)
elif positional_parameters_enforcement == POSITIONAL_WARNING:
logger.warning(message)
return wrapped(*args, **kwargs)
return positional_wrapper
if isinstance(max_positional_args, six.integer_types):
return positional_decorator
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\googleapiclient\http.py", line 920, in execute
)
self.body = parsed.query
self.headers["content-length"] = str(len(self.body))
# Handle retries for server-side errors.
resp, content = _retry_request(
http,
num_retries,
"request",
self._sleep,
self._rand,
File "C:[=11=]0 Projects\Applications\PY\flaskProjects\MDM\venv\Lib\site-packages\googleapiclient\http.py", line 191, in _retry_request
)
sleep(sleep_time)
try:
exception = None
resp, content = http.request(uri, method, *args, **kwargs)
# Retry on SSL errors and socket timeout errors.
except _ssl_SSLError as ssl_error:
exception = ssl_error
except socket.timeout as socket_timeout:
# Needs to be before socket.error as it's a subclass of OSError
AttributeError: 'str' object has no attribute 'request'
所以我发现问题是如何出现的：
有 3 个模块在执行相同的任务 google API 客户端 烧瓶要求 请求
这导致代码冲突。
将在我的 API 通话正常后更新