Instead of creating an api_call every time there is a call, create an api_callable object for each API method in the constructor, and pass the request and options to that api_callable. For example, existing implementation of publisher_api.create_topic:

        req = pubsub_pb2.Topic(name=name)
        settings = self._defaults['create_topic'].merge(options)
        api_call = api_callable.create_api_call(self.stub.CreateTopic,
                                                settings=settings)
        return api_call(req, metadata=self._headers)

Would change to:

        req = pubsub_pb2.Topic(name=name)
        return self._create_topic_callable(req, options)

New initialization code needs to be added to the constructor:

        self._create_topic_callable = api_callable.create(
            self.stub.CreateTopic,
            settings=defaults['create_topic'])

As for headers, they could be plumbed through defaults instead of passed individually to each api_callable.

As for page streaming, the api_callable would need to dynamically determine whether page streaming would be on or off based on the options.