chore(python): run blacken session for all directories with a noxfile (#162)

Source-Link: googleapis/synthtool@bc0de6e
Post-Processor: gcr.io/cloud-devrel-public-resources/owlbot-python:latest@sha256:39ad8c0570e4f5d2d3124a509de4fe975e799e2b97e0f58aed88f8880d5a8b60

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: gcf-owl-bot[bot]

scheduler/snippets/create_job.py

@@ -31,27 +31,20 @@ def create_scheduler_job(project_id, location_id, service_id):
 
     # Construct the request body.
     job = {
-        'app_engine_http_target': {
-            'app_engine_routing': {
-                'service': service_id
-            },
-            'relative_uri': '/log_payload',
-            'http_method': 1,
-            'body': 'Hello World'.encode()
+        "app_engine_http_target": {
+            "app_engine_routing": {"service": service_id},
+            "relative_uri": "/log_payload",
+            "http_method": 1,
+            "body": "Hello World".encode(),
         },
-        'schedule': '* * * * *',
-        'time_zone': 'America/Los_Angeles'
+        "schedule": "* * * * *",
+        "time_zone": "America/Los_Angeles",
     }
 
     # Use the client to send the job creation request.
-    response = client.create_job(
-        request={
-            "parent": parent,
-            "job": job
-        }
-    )
+    response = client.create_job(request={"parent": parent, "job": job})
 
-    print('Created job: {}'.format(response.name))
+    print("Created job: {}".format(response.name))
     # [END cloud_scheduler_create_job]
     return response
 

scheduler/snippets/create_job_test.py

@@ -16,18 +16,19 @@
 
 import create_job
 
-TEST_PROJECT_ID = os.getenv('GOOGLE_CLOUD_PROJECT')
-TEST_LOCATION = os.getenv('LOCATION_ID', 'us-central1')
+TEST_PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
+TEST_LOCATION = os.getenv("LOCATION_ID", "us-central1")
 
 
 def test_create_job(capsys):
     create_result = create_job.create_scheduler_job(
-        TEST_PROJECT_ID, TEST_LOCATION, 'my-service')
+        TEST_PROJECT_ID, TEST_LOCATION, "my-service"
+    )
     out, _ = capsys.readouterr()
-    assert 'Created job:' in out
+    assert "Created job:" in out
 
-    job_name = create_result.name.split('/')[-1]
+    job_name = create_result.name.split("/")[-1]
     create_job.delete_scheduler_job(TEST_PROJECT_ID, TEST_LOCATION, job_name)
 
     out, _ = capsys.readouterr()
-    assert 'Job deleted.' in out
+    assert "Job deleted." in out

scheduler/snippets/main.py

@@ -21,22 +21,24 @@
 
 
 # Define relative URI for job endpoint
-@app.route('/log_payload', methods=['POST'])
+@app.route("/log_payload", methods=["POST"])
 def example_task_handler():
     """Log the job payload."""
-    payload = request.get_data(as_text=True) or '(empty payload)'
-    print('Received job with payload: {}'.format(payload))
-    return 'Printed job payload: {}'.format(payload)
+    payload = request.get_data(as_text=True) or "(empty payload)"
+    print("Received job with payload: {}".format(payload))
+    return "Printed job payload: {}".format(payload)
+
+
 # [END cloud_scheduler_app]
 
 
-@app.route('/')
+@app.route("/")
 def hello():
     """Basic index to verify app is serving."""
-    return 'Hello World!'
+    return "Hello World!"
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     # This is used when running locally. Gunicorn is used to run the
     # application on Google App Engine. See entrypoint in app.yaml.
-    app.run(host='127.0.0.1', port=8080, debug=True)
+    app.run(host="127.0.0.1", port=8080, debug=True)

scheduler/snippets/main_test.py

@@ -18,28 +18,29 @@
 @pytest.fixture
 def app():
     import main
+
     main.app.testing = True
     return main.app.test_client()
 
 
 def test_index(app):
-    r = app.get('/')
+    r = app.get("/")
     assert r.status_code == 200
 
 
 def test_log_payload(capsys, app):
-    payload = 'test_payload'
+    payload = "test_payload"
 
-    r = app.post('/log_payload', data=payload)
+    r = app.post("/log_payload", data=payload)
     assert r.status_code == 200
 
     out, _ = capsys.readouterr()
     assert payload in out
 
 
 def test_empty_payload(capsys, app):
-    r = app.post('/log_payload')
+    r = app.post("/log_payload")
     assert r.status_code == 200
 
     out, _ = capsys.readouterr()
-    assert 'empty payload' in out
+    assert "empty payload" in out