Skip to main content
 首页 » 编程设计

python之Google Cloud Endpoints API 的 DRY 代码

2025年05月04日9小虾米

我想避免使用样板代码为我的 Google App Engine 应用程序的不同模型创建 Google Cloud Endpoints API。假设我有一个 Post、User 和 Category 模型。数据存储在数据存储中。我想使用资源 posts、users 和 categories 创建一个 REST API。我为 posts 资源编写了以下代码:

import endpoints 
from protorpc import messages 
from protorpc import message_types 
from protorpc import remote 
from blog.models import Post 
from cloud_endpoints import WEB_CLIENT_ID, ANDROID_CLIENT_ID, IOS_CLIENT_ID, ANDROID_AUDIENCE 
 
 
class PostMessage(messages.Message): 
    id = messages.StringField(1) 
    title = messages.StringField(2) 
    body = messages.StringField(3) 
 
 
class PostMessageCollection(messages.Message): 
    post_messages = messages.MessageField(PostMessage, 1, repeated=True) 
 
 
def post_to_message(post): 
    return PostMessage( 
        id=str(post.key()), 
        title=post.title, 
        body=post.body) 
 
 
ID_RESOURCE = endpoints.ResourceContainer( 
    message_types.VoidMessage, 
    id=messages.StringField(1, variant=messages.Variant.STRING)) 
 
PUT_RESOURCE = endpoints.ResourceContainer( 
    PostMessage, 
    id=messages.StringField(1, variant=messages.Variant.STRING)) 
 
POST_RESOURCE = endpoints.ResourceContainer(Post) 
 
 
@endpoints.api(name='posts', 
               version='v1', 
               allowed_client_ids=[WEB_CLIENT_ID, ANDROID_CLIENT_ID, IOS_CLIENT_ID], 
               audiences=[ANDROID_AUDIENCE]) 
class PostsApi(remote.Service): 
    """List""" 
    @endpoints.method(message_types.VoidMessage, 
                      PostMessageCollection, 
                      path='/posts', 
                      http_method='GET', 
                      name='posts.listPosts') 
    def list(self, unused_request): 
        post_messages = [] 
        for post in Post.all(): 
            post_messages.append(post_to_message(post)) 
 
        return PostCollection(post_messages=post_messages) 
 
    """Get""" 
    @endpoints.method(ID_RESOURCE, 
                      PostMessage, 
                      path='/posts/{id}', 
                      http_method='GET', 
                      name='posts.getPost') 
    def get(self, request): 
        try: 
            return post_to_message(Post.get(request.id)) 
 
        except (IndexError, TypeError): 
            raise endpoints.NotFoundException('Post %s not found.' % (request.id,)) 
 
    """Create""" 
    @endpoints.method(POST_RESOURCE, 
                      message_types.VoidMessage, 
                      path='/posts', 
                      http_method='POST', 
                      name='posts.createPost') 
    def create(self, request): 
        post = Post(title=request.title, body=request.body)\ 
        post.put() 
        return message_types.VoidMessage() 
 
    """Update""" 
    @endpoints.method(PUT_RESOURCE, 
                      message_types.VoidMessage, 
                      path='/posts/{id}', 
                      http_method='POST', 
                      name='posts.updatePost') 
    def update(self, request): 
        try: 
            post = Post.get(request.id) 
            post.title = request.title 
            post.body = request.body 
            return message_types.VoidMessage() 
        except (IndexError, TypeError): 
            raise endpoints.NotFoundException('Post %s not found.' % (request.id,)) 
 
    """Delete""" 
    @endpoints.method(ID_RESOURCE, 
                      message_types.VoidMessage, 
                      path='/posts/{id}', 
                      http_method='DELETE', 
                      name='posts.deletePost') 
    def delete(self, request): 
        try: 
            post = Post.get(request.id) 
            post.delete() 
            return message_types.VoidMessage() 
 
        except (IndexError, TypeError): 
            raise endpoints.NotFoundException('Post %s not found.' % (request.id,)) 

我可以复制/粘贴此代码并将“Post”更改为“Category”,然后编辑 PostMessage、PostMessageCollection 和 post_to_message,但这似乎是不好的做法。我不想重复自己。是否可以创建抽象 API 类并为 PostAPI、CategoryAPI 和 UserAPI 创建子类?或者是否有更好的方法来参数化 Post、PostMessage、PostMessageCollection、post_to_message 和资源路径 ( “/posts”、“/categories”和“/users”)这样我就不必为每个资源复制/粘贴类?这些类将具有相同的方法和相同的装饰器,我不想对每个资源重复这一点。我使用 Python 2.7。

请您参考如下方法:

我也偶然发现了同样的问题,不幸的是,这对于 google 云端点 是不可能的。方法装饰器需要请求描述(此处为 PostMessageCollection)。 message.Message 子类的请求描述不允许通过继承重用,因此所有消息类都必须完全定义,没有任何继承。

然而,您可以通过以下方式在某种程度上实现这一点(虽然我还没有测试过,但现在想到了 :)):

# All the message and response definitions have to be here, complete. 
 
class PostMessage(messages.Message): 
    id = messages.StringField(1) 
    title = messages.StringField(2) 
    body = messages.StringField(3) 
 
 
class PostMessageCollection(messages.Message): 
    post_messages = messages.MessageField(PostMessage, 1, repeated=True) 
 
 
def post_to_message(post): 
    return PostMessage( 
        id=str(post.key()), 
        title=post.title, 
        body=post.body) 
 
 
ID_RESOURCE = endpoints.ResourceContainer( 
    message_types.VoidMessage, 
    id=messages.StringField(1, variant=messages.Variant.STRING)) 
 
PUT_RESOURCE = endpoints.ResourceContainer( 
    PostMessage, 
    id=messages.StringField(1, variant=messages.Variant.STRING)) 
 
POST_RESOURCE = endpoints.ResourceContainer(Post) 
 
# Now define all the 'Category' related messages here. 
 
 
@endpoints.api(name='posts_n_categories',  # The name can be a common one. 
               version='v1', 
               allowed_client_ids=[WEB_CLIENT_ID, ANDROID_CLIENT_ID, IOS_CLIENT_ID], 
               audiences=[ANDROID_AUDIENCE]) 
class BaseAPI(remote.Service): 
    """List""" 
    # Common defs go here. 
    MessageCollection = messages.Message 
    PATH = '/' 
    NAME = '' 
 
    @staticmethod 
    def converter(x): 
        raise NotImplemented 
 
    iterator = [] 
    collection = messages.Message 
    @endpoints.method(message_types.VoidMessage, 
                      MessageCollection, 
                      path=PATH, 
                      http_method='GET', 
                      name=NAME) 
    def list(self, unused_request): 
        # Do the common work here. You can  
        _messages = [] 
        for post in self.__class__.iterator.all(): 
            _messages.append(self.__class__.converter(post)) 
 
        return self.__class__.collection(post_messages=_messages) 
 
 
 
 
@endpoints.api(name='posts',  # The name can be different. 
               version='v1', 
               allowed_client_ids=[WEB_CLIENT_ID, ANDROID_CLIENT_ID, IOS_CLIENT_ID], 
               audiences=[ANDROID_AUDIENCE]) 
class PostAPI(Base): 
    # Post specific defs go here. 
    MessageCollection = PostMessageCollection 
    PATH = '/posts' 
    NAME = 'posts.listPosts' 
    converter = post_to_message 
    iterator = Post 
    collection = PostCollection 
 
 
# Define the category class here. 

显然,它并没有节省多少时间。