title: Unable to publish Pubsub message in Airflow Python 3
body: <p>I am unable to publish using the <code>PubSubHook</code> in Airflow with Python 3. Everything works perfectly with Python 2, but with Python 3 I get this error <code>{models.py:1760} ERROR - Object of type 'bytes' is not JSON serializable</code>. It seems that encoding the message in Python 3 results in a byte that then can't be handled by the JSON serializer.     </p>
  
  <p>The following works fine in Python 2:</p>
  
  <pre><code>def send_message_to_pubsub(message):
      pubsub_message = {'data': b64encode(message)}
      hook = PubSubHook(gcp_conn_id='google_cloud_default')
      hook.publish('project-name', 'topic-name', [pubsub_message])
  </code></pre>
  
  <p>The example <a href="https://github.com/apache/airflow/blob/master/airflow/contrib/example_dags/example_pubsub_flow.py" rel="nofollow noreferrer">here</a> doesn't work as well with Python 3. </p>
  
  <p><strong>Update 1:</strong></p>
  
  <p>Tried with the following but got error:</p>
  
  <pre><code>def send_message_to_pubsub():
      message = 'Test message'
      pubsub_message = {'data': b64encode(message).decode()}
      hook = PubSubHook(gcp_conn_id='google_cloud_default')
      hook.publish('project-name', 'topic-name', [pubsub_message])
  
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test [2019-03-18 17:10:28,903] {models.py:1760} ERROR - a bytes-like object is required, not 'str'
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test Traceback (most recent call last):
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test   File "/usr/local/lib/airflow/airflow/models.py", line 1659, in _run_raw_task
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test     result = task_copy.execute(context=context)
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test   File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 95, in execute
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test     return_value = self.execute_callable()
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test   File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 100, in execute_callable
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test     return self.python_callable(*self.op_args, **self.op_kwargs)
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test   File "/home/airflow/gcs/dags/pubsub-test-dag.py", line 31, in send_message_to_pubsub
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test     pubsub_message = {'data': b64encode(message).decode()}
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test   File "/opt/python3.6/lib/python3.6/base64.py", line 58, in b64encode
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test     encoded = binascii.b2a_base64(s, newline=False)
  {base_task_runner.py:101} INFO - Job 1962: Subtask pub_sub_test TypeError: a bytes-like object is required, not 'str'
  </code></pre>
  
  <p><strong>Update 2:</strong></p>
  
  <p>Tried with the following, resulted in a different error. This time from JSON serializer:</p>
  
  <pre><code>def send_message_to_pubsub():
      message = 'Test message'
      pubsub_message = {'data': b64encode(message.encode())}
      hook = PubSubHook(gcp_conn_id='google_cloud_default')
      hook.publish('project', 'topic', [pubsub_message]) 
  
  [2019-03-19 10:44:29,845] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test [2019-03-19 10:44:29,841] {models.py:1760} ERROR - Object of type 'bytes' is not JSON serializable
  [2019-03-19 10:44:29,846] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test Traceback (most recent call last):
  [2019-03-19 10:44:29,846] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/usr/local/lib/airflow/airflow/models.py", line 1659, in _run_raw_task
  [2019-03-19 10:44:29,847] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     result = task_copy.execute(context=context)
  [2019-03-19 10:44:29,847] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 95, in execute
  [2019-03-19 10:44:29,847] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     return_value = self.execute_callable()
  [2019-03-19 10:44:29,847] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 100, in execute_callable
  [2019-03-19 10:44:29,848] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     return self.python_callable(*self.op_args, **self.op_kwargs)
  [2019-03-19 10:44:29,848] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/home/airflow/gcs/dags/pubsub-test-dag.py", line 33, in send_message_to_pubsub
  [2019-03-19 10:44:29,848] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     hook.publish('project', 'topic', [pubsub_message])
  [2019-03-19 10:44:29,848] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/usr/local/lib/airflow/airflow/contrib/hooks/gcp_pubsub_hook.py", line 75, in publish
  [2019-03-19 10:44:29,849] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     topic=full_topic, body=body)
  [2019-03-19 10:44:29,849] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/opt/python3.6/lib/python3.6/site-packages/googleapiclient/discovery.py", line 795, in method
  [2019-03-19 10:44:29,849] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     actual_path_params, actual_query_params, body_value)
  [2019-03-19 10:44:29,850] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/opt/python3.6/lib/python3.6/site-packages/googleapiclient/model.py", line 151, in request
  [2019-03-19 10:44:29,850] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     body_value = self.serialize(body_value)
  [2019-03-19 10:44:29,850] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/opt/python3.6/lib/python3.6/site-packages/googleapiclient/model.py", line 260, in serialize
  [2019-03-19 10:44:29,850] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     return json.dumps(body_value)
  [2019-03-19 10:44:29,851] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/opt/python3.6/lib/python3.6/json/__init__.py", line 231, in dumps
  [2019-03-19 10:44:29,851] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     return _default_encoder.encode(obj)
  [2019-03-19 10:44:29,853] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/opt/python3.6/lib/python3.6/json/encoder.py", line 199, in encode
  [2019-03-19 10:44:29,853] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     chunks = self.iterencode(o, _one_shot=True)
  [2019-03-19 10:44:29,853] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test   File "/opt/python3.6/lib/python3.6/json/encoder.py", line 257, in iterencode
  [2019-03-19 10:44:29,854] {base_task_runner.py:101} INFO - Job 2172: Subtask pub_sub_test     return _iterencode(o, 0)
  [2019-03-19 10:44:29,852] {models.py:1791} INFO - Marking task as FAILED.
  </code></pre>