google cloud pubsub ImportError：无法导入名称类型

Posted 2023-02-16

【中文标题】google cloud pubsub ImportError：无法导入名称类型

【英文标题】：google cloud pubsub ImportError: cannot import name types

【发布时间】：2018-12-11 07:14:05

【问题描述】：

我使用 google-cloud-pubsub 为标准环境在 python 上为 Google appengine 编写了一个小程序。我收到一个错误 ImportError：无法导入名称类型。我还看到problem 仍然没有解决。但也许有人在标准环境中创建了一个 sub-pub？

我安装 lib：pip installall -t lib google-cloud-pubsub。

在 appengine_config.py 中：vendor.add('lib')

访问 appengine 应用程序时出错：

Traceback (most recent call last):
  File "/base/alloc/tmpfs/dynamic_runtimes/python27g/7894e0c59273b2b7/python27/python27_lib/versions/1/google/appengine/runtime/wsgi.py", line 240, in Handle
    handler = _config_handle.add_wsgi_middleware(self._LoadHandler())
  File "/base/alloc/tmpfs/dynamic_runtimes/python27g/7894e0c59273b2b7/python27/python27_lib/versions/1/google/appengine/runtime/wsgi.py", line 299, in _LoadHandler
    handler, path, err = LoadObject(self._handler)
  File "/base/alloc/tmpfs/dynamic_runtimes/python27g/7894e0c59273b2b7/python27/python27_lib/versions/1/google/appengine/runtime/wsgi.py", line 85, in LoadObject
    obj = __import__(path[0])
  File "/base/data/home/apps/e~pqcloud-sp/agg2:20180703t012034.410851540172441919/service_main.py", line 5, in <module>
    from google.cloud.pubsub_v1 import PublisherClient
  File "/base/data/home/apps/e~pqcloud-sp/agg2:20180703t012034.410851540172441919/lib/google/cloud/pubsub_v1/__init__.py", line 17, in <module>
    from google.cloud.pubsub_v1 import types
ImportError: cannot import name types

app.yaml：

runtime: python27
api_version: 1
threadsafe: true
service: agg2

handlers:
- url: .*
  script: service_main.app

libraries:
- name: webapp2
  version: "2.5.1"
- name: jinja2
  version: latest

skip_files:
- ^(.*/)?#.*#$
- ^(.*/)?.*~$
- ^(.*/)?.*\.py[co]$
- ^(.*/)?.*/RCS/.*$
- ^(.*/)?\..*$

service_main.py：

import os
import logging
import webapp2

from google.cloud.pubsub_v1 import PublisherClient

logger = logging.getLogger('service_main')
logger.setLevel(logging.WARNING)

class ServiceTaskMainHandler(webapp2.RequestHandler):
    def get(self):
        self.response.headers['Content-Type'] = 'application/json'
        self.response.out.write('test')


config = 
    'webapp2_extras.sessions': 
        'secret_key': 'YOUR_SECRET_KEY'
    


MAIN_ROUTE = [
    webapp2.Route('/', ServiceTaskMainHandler, name='main'),
]

app = webapp2.WSGIApplication(MAIN_ROUTE, debug=True, config=config)

---

树库/google/cloud/pubsub_v1

lib/google/cloud/pubsub_v1
├── exceptions.py
├── exceptions.pyc
├── futures.py
├── futures.pyc
├── gapic
│   ├── __init__.py
│   ├── __init__.pyc
│   ├── publisher_client_config.py
│   ├── publisher_client_config.pyc
│   ├── publisher_client.py
│   ├── publisher_client.pyc
│   ├── subscriber_client_config.py
│   ├── subscriber_client_config.pyc
│   ├── subscriber_client.py
│   └── subscriber_client.pyc
├── _gapic.py
├── _gapic.pyc
├── __init__.py
├── __init__.pyc
├── proto
│   ├── __init__.py
│   ├── __init__.pyc
│   ├── pubsub_pb2_grpc.py
│   ├── pubsub_pb2_grpc.pyc
│   ├── pubsub_pb2.py
│   └── pubsub_pb2.pyc
├── publisher
│   ├── batch
│   │   ├── base.py
│   │   ├── base.pyc
│   │   ├── __init__.py
│   │   ├── __init__.pyc
│   │   ├── thread.py
│   │   └── thread.pyc
│   ├── client.py
│   ├── client.pyc
│   ├── exceptions.py
│   ├── exceptions.pyc
│   ├── futures.py
│   ├── futures.pyc
│   ├── __init__.py
│   └── __init__.pyc
├── subscriber
│   ├── client.py
│   ├── client.pyc
│   ├── futures.py
│   ├── futures.pyc
│   ├── __init__.py
│   ├── __init__.pyc
│   ├── message.py
│   ├── message.pyc
│   ├── _protocol
│   │   ├── bidi.py
│   │   ├── bidi.pyc
│   │   ├── dispatcher.py
│   │   ├── dispatcher.pyc
│   │   ├── heartbeater.py
│   │   ├── heartbeater.pyc
│   │   ├── helper_threads.py
│   │   ├── helper_threads.pyc
│   │   ├── histogram.py
│   │   ├── histogram.pyc
│   │   ├── __init__.py
│   │   ├── __init__.pyc
│   │   ├── leaser.py
│   │   ├── leaser.pyc
│   │   ├── requests.py
│   │   ├── requests.pyc
│   │   ├── streaming_pull_manager.py
│   │   └── streaming_pull_manager.pyc
│   ├── scheduler.py
│   └── scheduler.pyc
├── types.py
└── types.pyc

【参考方案1】：

您的代码失败的原因是 App Engine Standard 的 Python2.7 运行时不支持 Pub/Sub Cloud 客户端库，仅支持 Pub/Sub API 客户端库。 some new code samples 展示了如何将 Pub/Sub 与 App Engine Standard 结合使用。

import googleapiclient.discovery
import base64

service = build('pubsub', 'v1')

topic_path = 'projects/your_project_id/topics/your_topic'

service.projects().topics().publish(
    topic=topic_path, body=
      "messages": [
          "data": base64.b64encode(data)
      ]
    ).execute()

更新：GAE (Google App Engine) Standard 和 GAE Flexible Python 3 Runtime 都支持 Cloud Pub/Sub 客户端库。