fixit: Clean up Python sample at cloud-sql/mysql/sqlalchemy

cloud-sql/mysql/sqlalchemy/app.py

@@ -33,6 +33,7 @@
 
 
 def init_connection_pool() -> sqlalchemy.engine.base.Engine:
+    """Sets up connection pool for the app."""
     # use a TCP socket when INSTANCE_HOST (e.g. 127.0.0.1) is defined
     if os.environ.get("INSTANCE_HOST"):
         return connect_tcp_socket()
@@ -45,7 +46,11 @@ def init_connection_pool() -> sqlalchemy.engine.base.Engine:
     if os.environ.get("INSTANCE_CONNECTION_NAME"):
         # Either a DB_USER or a DB_IAM_USER should be defined. If both are
         # defined, DB_IAM_USER takes precedence.
-        return connect_with_connector_auto_iam_authn() if os.environ.get("DB_IAM_USER") else connect_with_connector()
+        return (
+            connect_with_connector_auto_iam_authn()
+            if os.environ.get("DB_IAM_USER")
+            else connect_with_connector()
+        )
 
     raise ValueError(
         "Missing database connection type. Please define one of INSTANCE_HOST, INSTANCE_UNIX_SOCKET, or INSTANCE_CONNECTION_NAME"
@@ -54,12 +59,15 @@ def init_connection_pool() -> sqlalchemy.engine.base.Engine:
 
 # create 'votes' table in database if it does not already exist
 def migrate_db(db: sqlalchemy.engine.base.Engine) -> None:
+    """Creates the `votes` table if it doesn't exist."""
     with db.connect() as conn:
-        conn.execute(sqlalchemy.text(
-            "CREATE TABLE IF NOT EXISTS votes "
-            "( vote_id SERIAL NOT NULL, time_cast timestamp NOT NULL, "
-            "candidate VARCHAR(6) NOT NULL, PRIMARY KEY (vote_id) );"
-        ))
+        conn.execute(
+            sqlalchemy.text(
+                "CREATE TABLE IF NOT EXISTS votes "
+                "( vote_id SERIAL NOT NULL, time_cast timestamp NOT NULL, "
+                "candidate VARCHAR(6) NOT NULL, PRIMARY KEY (vote_id) );"
+            )
+        )
         conn.commit()
 
 
@@ -75,32 +83,45 @@ def migrate_db(db: sqlalchemy.engine.base.Engine) -> None:
 # as the function is loaded. This is primarily to help testing.
 @app.before_first_request
 def init_db() -> sqlalchemy.engine.base.Engine:
+    """Initiates connection to database and its' structure."""
     global db
     db = init_connection_pool()
     migrate_db(db)
 
 
 @app.route("/", methods=["GET"])
 def render_index() -> str:
+    """Serves the index page of the app."""
     context = get_index_context(db)
     return render_template("index.html", **context)
 
 
 @app.route("/votes", methods=["POST"])
 def cast_vote() -> Response:
-    team = request.form['team']
+    """Processes a single vote from user."""
+    team = request.form["team"]
     return save_vote(db, team)
 
 
 # get_index_context gets data required for rendering HTML application
 def get_index_context(db: sqlalchemy.engine.base.Engine) -> dict:
+    """Retrieves data from the database about the votes.
+
+    Args:
+        db: Connection to the database.
+
+    Returns:
+        A dictionary containing information about votes.
+    """
     votes = []
 
     with db.connect() as conn:
         # Execute the query and fetch all results
-        recent_votes = conn.execute(sqlalchemy.text(
-            "SELECT candidate, time_cast FROM votes ORDER BY time_cast DESC LIMIT 5"
-        )).fetchall()
+        recent_votes = conn.execute(
+            sqlalchemy.text(
+                "SELECT candidate, time_cast FROM votes ORDER BY time_cast DESC LIMIT 5"
+            )
+        ).fetchall()
         # Convert the results into a list of dicts representing votes
         for row in recent_votes:
             votes.append({"candidate": row[0], "time_cast": row[1]})
@@ -122,6 +143,15 @@ def get_index_context(db: sqlalchemy.engine.base.Engine) -> dict:
 
 # save_vote saves a vote to the database that was retrieved from form data
 def save_vote(db: sqlalchemy.engine.base.Engine, team: str) -> Response:
+    """Saves a single vote into the database.
+
+    Args:
+        db: Connection to the database.
+        team: The identifier of a team the vote is casted on.
+
+    Returns:
+        A HTTP response that can be sent to the client.
+    """
     time_cast = datetime.datetime.now(tz=datetime.timezone.utc)
     # Verify that the team is one of the allowed options
     if team != "TABS" and team != "SPACES":

cloud-sql/mysql/sqlalchemy/connect_connector.py

@@ -32,7 +32,9 @@ def connect_with_connector() -> sqlalchemy.engine.base.Engine:
     # Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
     # keep secrets safe.
 
-    instance_connection_name = os.environ["INSTANCE_CONNECTION_NAME"]  # e.g. 'project:region:instance'
+    instance_connection_name = os.environ[
+        "INSTANCE_CONNECTION_NAME"
+    ]  # e.g. 'project:region:instance'
     db_user = os.environ["DB_USER"]  # e.g. 'my-db-user'
     db_pass = os.environ["DB_PASS"]  # e.g. 'my-db-password'
     db_name = os.environ["DB_NAME"]  # e.g. 'my-database'
@@ -57,18 +59,14 @@ def getconn() -> pymysql.connections.Connection:
         # [START_EXCLUDE]
         # Pool size is the maximum number of permanent connections to keep.
         pool_size=5,
-
         # Temporarily exceeds the set pool_size if no connections are available.
         max_overflow=2,
-
         # The total number of concurrent connections for your application will be
         # a total of pool_size and max_overflow.
-
         # 'pool_timeout' is the maximum number of seconds to wait when retrieving a
         # new connection from the pool. After the specified amount of time, an
         # exception will be thrown.
         pool_timeout=30,  # 30 seconds
-
         # 'pool_recycle' is the maximum number of seconds a connection can persist.
         # Connections that live longer than the specified amount of time will be
         # re-established
@@ -77,4 +75,5 @@ def getconn() -> pymysql.connections.Connection:
     )
     return pool
 
+
 # [END cloud_sql_mysql_sqlalchemy_connect_connector]

cloud-sql/mysql/sqlalchemy/connect_connector_auto_iam_authn.py

@@ -31,7 +31,9 @@ def connect_with_connector_auto_iam_authn() -> sqlalchemy.engine.base.Engine:
     # secure - consider a more secure solution such as
     # Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
     # keep secrets safe.
-    instance_connection_name = os.environ["INSTANCE_CONNECTION_NAME"]  # e.g. 'project:region:instance'
+    instance_connection_name = os.environ[
+        "INSTANCE_CONNECTION_NAME"
+    ]  # e.g. 'project:region:instance'
     db_iam_user = os.environ["DB_IAM_USER"]  # e.g. 'service-account-name'
     db_name = os.environ["DB_NAME"]  # e.g. 'my-database'
 
@@ -59,18 +61,14 @@ def getconn() -> pymysql.connections.Connection:
         # [START_EXCLUDE]
         # Pool size is the maximum number of permanent connections to keep.
         pool_size=5,
-
         # Temporarily exceeds the set pool_size if no connections are available.
         max_overflow=2,
-
         # The total number of concurrent connections for your application will be
         # a total of pool_size and max_overflow.
-
         # 'pool_timeout' is the maximum number of seconds to wait when retrieving a
         # new connection from the pool. After the specified amount of time, an
         # exception will be thrown.
         pool_timeout=30,  # 30 seconds
-
         # 'pool_recycle' is the maximum number of seconds a connection can persist.
         # Connections that live longer than the specified amount of time will be
         # re-established
@@ -79,4 +77,5 @@ def getconn() -> pymysql.connections.Connection:
     )
     return pool
 
+
 # [END cloud_sql_mysql_sqlalchemy_auto_iam_authn]

cloud-sql/mysql/sqlalchemy/connect_tcp.py

@@ -22,12 +22,14 @@
 
 
 def connect_tcp_socket() -> sqlalchemy.engine.base.Engine:
-    """ Initializes a TCP connection pool for a Cloud SQL instance of MySQL. """
+    """Initializes a TCP connection pool for a Cloud SQL instance of MySQL."""
     # Note: Saving credentials in environment variables is convenient, but not
     # secure - consider a more secure solution such as
     # Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
     # keep secrets safe.
-    db_host = os.environ["INSTANCE_HOST"]  # e.g. '127.0.0.1' ('172.17.0.1' if deployed to GAE Flex)
+    db_host = os.environ[
+        "INSTANCE_HOST"
+    ]  # e.g. '127.0.0.1' ('172.17.0.1' if deployed to GAE Flex)
     db_user = os.environ["DB_USER"]  # e.g. 'my-db-user'
     db_pass = os.environ["DB_PASS"]  # e.g. 'my-db-password'
     db_name = os.environ["DB_NAME"]  # e.g. 'my-database'
@@ -43,11 +45,7 @@ def connect_tcp_socket() -> sqlalchemy.engine.base.Engine:
         db_cert = os.environ["DB_CERT"]  # e.g. '/path/to/my/client-cert.pem'
         db_key = os.environ["DB_KEY"]  # e.g. '/path/to/my/client-key.pem'
 
-        ssl_args = {
-            "ssl_ca": db_root_cert,
-            "ssl_cert": db_cert,
-            "ssl_key": db_key
-        }
+        ssl_args = {"ssl_ca": db_root_cert, "ssl_cert": db_cert, "ssl_key": db_key}
         connect_args = ssl_args
 
     # [START cloud_sql_mysql_sqlalchemy_connect_tcp]
@@ -74,19 +72,16 @@ def connect_tcp_socket() -> sqlalchemy.engine.base.Engine:
         # The total number of concurrent connections for your application will be
         # a total of pool_size and max_overflow.
         # [END cloud_sql_mysql_sqlalchemy_limit]
-
         # [START cloud_sql_mysql_sqlalchemy_backoff]
         # SQLAlchemy automatically uses delays between failed connection attempts,
         # but provides no arguments for configuration.
         # [END cloud_sql_mysql_sqlalchemy_backoff]
-
         # [START cloud_sql_mysql_sqlalchemy_timeout]
         # 'pool_timeout' is the maximum number of seconds to wait when retrieving a
         # new connection from the pool. After the specified amount of time, an
         # exception will be thrown.
         pool_timeout=30,  # 30 seconds
         # [END cloud_sql_mysql_sqlalchemy_timeout]
-
         # [START cloud_sql_mysql_sqlalchemy_lifetime]
         # 'pool_recycle' is the maximum number of seconds a connection can persist.
         # Connections that live longer than the specified amount of time will be
@@ -97,6 +92,7 @@ def connect_tcp_socket() -> sqlalchemy.engine.base.Engine:
     )
     return pool
 
+
 # [END cloud_sql_mysql_sqlalchemy_connect_tcp_sslcerts]
 # [END cloud_sql_mysql_sqlalchemy_sslcerts]
 # [END cloud_sql_mysql_sqlalchemy_connect_tcp]

cloud-sql/mysql/sqlalchemy/connect_unix.py

@@ -19,15 +19,17 @@
 
 
 def connect_unix_socket() -> sqlalchemy.engine.base.Engine:
-    """ Initializes a Unix socket connection pool for a Cloud SQL instance of MySQL. """
+    """Initializes a Unix socket connection pool for a Cloud SQL instance of MySQL."""
     # Note: Saving credentials in environment variables is convenient, but not
     # secure - consider a more secure solution such as
     # Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
     # keep secrets safe.
     db_user = os.environ["DB_USER"]  # e.g. 'my-database-user'
     db_pass = os.environ["DB_PASS"]  # e.g. 'my-database-password'
     db_name = os.environ["DB_NAME"]  # e.g. 'my-database'
-    unix_socket_path = os.environ["INSTANCE_UNIX_SOCKET"]  # e.g. '/cloudsql/project:region:instance'
+    unix_socket_path = os.environ[
+        "INSTANCE_UNIX_SOCKET"
+    ]  # e.g. '/cloudsql/project:region:instance'
 
     pool = sqlalchemy.create_engine(
         # Equivalent URL:
@@ -42,18 +44,14 @@ def connect_unix_socket() -> sqlalchemy.engine.base.Engine:
         # [START_EXCLUDE]
         # Pool size is the maximum number of permanent connections to keep.
         pool_size=5,
-
         # Temporarily exceeds the set pool_size if no connections are available.
         max_overflow=2,
-
         # The total number of concurrent connections for your application will be
         # a total of pool_size and max_overflow.
-
         # 'pool_timeout' is the maximum number of seconds to wait when retrieving a
         # new connection from the pool. After the specified amount of time, an
         # exception will be thrown.
         pool_timeout=30,  # 30 seconds
-
         # 'pool_recycle' is the maximum number of seconds a connection can persist.
         # Connections that live longer than the specified amount of time will be
         # re-established
@@ -62,4 +60,5 @@ def connect_unix_socket() -> sqlalchemy.engine.base.Engine:
     )
     return pool
 
+
 # [END cloud_sql_mysql_sqlalchemy_connect_unix]

cloud-sql/mysql/sqlalchemy/main.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from flask import render_template, Response
+from flask import render_template, Request, Response
 
 import functions_framework
 
@@ -27,7 +27,15 @@
 
 
 @functions_framework.http
-def votes(request):
+def votes(request: Request) -> Response:
+    """Handles HTTP requests to our application.
+
+    Args:
+        request: a Flask request object.
+
+    Returns:
+        Flask HTTP Response to the client.
+    """
     if request.method == "GET":
         context = get_index_context(db)
         return render_template("index.html", **context)