Easy to use Google Pub/Sub

This relates to #114 & #119

This makes makes all config variables nullable falling back to standard google envars, without breaking the current api.

The new apis would look like this is you have GOOGLE_APPLICATION_CREDENTIALS set.

rele.config.setup()

w = Worker([sub1, sub2])
w.run_forever()

TBH, I think reading global configs from the environment is easier to reason about than a singleton so I'd suggest that. Any non globals should just be passed into the instances on init.

:tophat: What?

Provide a description of what has been implemented.

:thinking: Why?

Give an explanation of why.

:link: Related issue

Add related issue's number. Example: Fix #1

I'm getting the following error when trying to run rele.config.setup using default credentials:

rele.config.setup({
	"GC_CREDENTIALS_PATH": None,
    "MIDDLEWARE": [
     	"rele.contrib.LoggingMiddleware",
        "rele.contrib.FlaskMiddleware",
     ],
     "APP_NAME": "smart_comms_planner",
}, flask_app=app)

output:

  File "/Users/matthewbridges/repos/smart-comms-planner/src/__init__.py", line 66, in <module>
    rele.config.setup(settings["rele"], flask_app=app)
  File "/Users/matthewbridges/.local/share/virtualenvs/smart-comms-planner-yrxYHqso/lib/python3.8/site-packages/rele/config.py", line 69, in setup
    init_global_publisher(config)
  File "/Users/matthewbridges/.local/share/virtualenvs/smart-comms-planner-yrxYHqso/lib/python3.8/site-packages/rele/publishing.py", line 10, in init_global_publisher
    gc_project_id=config.gc_project_id,
  File "/Users/matthewbridges/.local/share/virtualenvs/smart-comms-planner-yrxYHqso/lib/python3.8/site-packages/rele/config.py", line 59, in gc_project_id
    return self.credentials.project_id
AttributeError: 'Credentials' object has no attribute 'project_id'

The code is incorrectly attempting to read project_id off the credentials object, when in fact it is returned as a tuple from get_google_defaults.

I've added a proposed fix here: https://github.com/mercadona/rele/pull/195

We currently have flake8 in our linting command. We do not have isort in the project style guidelines.

Shall we add it?

Or try out something new like black?

I would like to be able to configure the timeout when publishing a message in a blocking fashion. Right now, the default and hard coded way is set to 3.0 seconds.

I propose adding a configuration so I can declare any number of seconds for the publisher.

Ex.

RELE = {
	...
	'PUBLISHER_TIMEOUT': 5.0
}

I'm adding rele to flask app, and the subscription callbacks need the flask app_context. I've done this by way of middleware:

eg:

class FlaskMiddleware(BaseMiddleware):
    def pre_process_message(self, subscription, message):
        from server import app
        self.ctx = app.app_context()
        self.ctx.push()

    def post_process_message(self):
        self.ctx.pop()

But to make this reusable across our services and other flask apps, I'd need a way to add arbitrary data to the config that is passed to middleware.setup method or have an easy way to call custom middleware functions.

eg:

class FlaskMiddleware(BaseMiddleware):
    def setup(self, config):
        self.app = config["FLASK_APP"]

    def pre_process_message(self, subscription, message):
        self.ctx = self.app.app_context()
        self.ctx.push()

    def post_process_message(self):
        self.ctx.pop()

One consistent peice of feedback that I have received is simplifying the Worker class. Right now, you must Initialize with the subs, each individual config attribute, run setup, run start, and then sleep.

Like this:

worker = Worker(
    [photo_uploaded],
    config.gc_project_id,
    config.credentials,
    config.ack_deadline,
)
worker.setup()
worker.start()
sleep(120)

I propose we simplify the API to run a worker to look something like this instead:

worker = Worker([photo_uploaded], config)
worker.run(sleep=120)

run would call both setup and start, and we could add the standard sleep method. In addition we consolidate the configuration into one attribute.

This would be backwards compatible since we will be creating a new method. And the change to Worker initialization could also fall back to the declared attributes if defined. Otherwise, use the config object.

The sub decorator can get some ease of use improvements

1. Use functools.wraps to preserve __name__ and __doc__ of the original function
2. Inspect callback function signature and raise an exception if the signature is not compatible (ie. missing data)
3. Log a warning message if the function cannot be discovered (ie. no sub in the path)

Slightly related to the second point: @sub("topic", filter_by=42) will raise a TypeError at runtime (because filters are used as callables but they are not checked)

In the 1.0 release, I had to pin pubsub to <2.0. Otherwise, our tests would break. It seems there were major changes, and we need to update some usages in our code.

To reproduce:

1. update google pubsub library to >2.0.
2. Run tests

Hey 👋 first of all thanks for this great library!

This might not be a strictly Relé related issue, but when I tried to pause execution to debug some issue (ie dropping a __import__("pdb").set_trace()) inside the subscription message handler function it did not stop, but just kept on processing messages.

This might be a trivial problem to overcome, but my search engine fu failed me. I'm guessing the problem is that the subscriber handles the message in a thread, so it would somehow need to let the parent know to stop, but not really sure how to go about it.

This is useful if you know you will never be able to handle the message and don't have a dead-letter queue setup.

:tophat: What?

A custom exception and middleware to automate and ignore bad messages.

:thinking: Why?

I've been using this a fair amount in some of our services and thought it may be useful to others.

:tophat: What?

Add python 3.8 to travis.

:thinking: Why?

It was officially released on Oct 14, 2019

This would allow us to officially add Python 3.8. However, it seems that grpcio is not compiled for Python 3.8 and therefore the travis build takes much longer to complete.. Ran for 5 min 40 sec.

So either we can wait and not support 3.8 yet, or merge this and have the travis build take longer than expected.

I leave it as a draft until we decide.

:tophat: What?

Remove PostgreSQL service from Travis pipeline

:thinking: Why?

• It's not used
• It could give a false sensation of security by believing there are tests checking database features (Django DB)

:tophat: What?

• Run the test suite against the minimumPython supported (3.6) and the maximum version supported (3.10)
• Also update the dist image to bionic because it's the only one with support for all the Python versions.
• Add the PIP classifiers for Python 3.10

:thinking: Why?

Be sure the library is compatible with the latest/current version of Python (3.10)

Currently, async functions can't be used with the @sub context manager.

@sub(topic='my-topic'):
async def async_handler(data, **kwargs):
    print('data', data)

If you try, you'll get the following error.

Configuring worker with 1 subscription(s)...
  dicom_uploads - handle_upload
/usr/local/lib/python3.9/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py:126: RuntimeWarning: coroutine 'handle_upload' was never awaited
  callback(message)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

There are some simple workarounds, but it would be nice if this was supported natively.

@sub(topic='my-topic'):
def sync_handler(data, **kwargs)
    return asyncio.run(async_handler(data, **kwargs))

async def async_handler(data, **kwargs):
    print('data', data)

:tophat: What?

Allow custom deserializers other than the default json deserializer.

:thinking: Why?

Some of GCP's pub/sub notifications have messages that are simple strings which can't be deserialized as json.

For example, notifications that a record was inserted into the dicom store comes through with a string indicating the path to that record and nothing more.

:link: https://github.com/mercadona/rele/issues/229

Currently rele's subscription model only supports consuming messages that are json deserializable.

Unfortunately some of GCP's core services publish non-json messages. For example, the gcp dicom store publishes notifications to a topic when a new dicom instance was inserted, but the message field is just the dicom instance path. When attempting to consume these messages through rele, we get the following error.

Exception raised while processing message for dicom_uploads - handle_upload: JSONDecodeError
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/rele/subscription.py", line 112, in __call__
    data = json.loads(message.data.decode("utf-8"))
  File "/usr/local/lib/python3.9/json/__init__.py", line 346, in loads
    return _default_decoder.decode(s)
  File "/usr/local/lib/python3.9/json/decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/local/lib/python3.9/json/decoder.py", line 355, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

Can we add an option to disable deserialization of these messages? Something like:

from rele import sub

@sub(topic='my-topic', deserialize_message=False):
def handle_upload(data, **kwargs):
    print('Handling data', data)