def connect_tcp_socket()

in chaostoolkit-examples/cloudsql-cloudrun-pbr-fault-injection/app/cloudruncode/main.py [0:0]

• 43 lines of code
• 2 McCabe index (conditional complexity)

def connect_tcp_socket() -> sqlalchemy.engine.base.Engine:

    """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.get("db_host")
    db_name = "mysql" 
    db_port = 3306 

    # IP lookup from hostname
    print(f'The {db_host} IP Address is {socket.gethostbyname(db_host)}')
    print("2------------------")
    project_id = os.environ.get("project_id")
    
    # Build the resource name of the secret version.

    pool = sqlalchemy.create_engine(
        # Equivalent URL:
        # mysql+pymysql://<db_user>:<db_pass>@<db_host>:<db_port>/<db_name>
        sqlalchemy.engine.url.URL.create(
            drivername="mysql+pymysql",
            username=get_secret(f"projects/{project_id}/secrets/db-user-pbr/versions/1"),
            password=get_secret(f"projects/{project_id}/secrets/db-password-pbr/versions/1"),
            host=db_host,
            port=db_port
        ),
        # ...
    )
    res={}
    with pool.connect() as db_conn:

    # query and fetch ratings table
        try:
            res= db_conn.execute(sqlalchemy.text("CREATE DATABASE testdb;"))
        except:
            print("database already created")
        try:
            res= db_conn.execute(sqlalchemy.text("USE testdb;"))
        except:
            print("already in database")
        try:
            res= db_conn.execute(sqlalchemy.text("CREATE TABLE employeetable (name VARCHAR(255), id INT)"))
        except:
            print("Already table is there")
        try:
            res= db_conn.execute(sqlalchemy.text("INSERT INTO employeetable (name, id) VALUES ('Peter', 1)"))
        except:
            print("Already data is present")
        try:
            res= db_conn.execute(sqlalchemy.text("INSERT INTO employeetable (name, id) VALUES ('John', 2)"))
        except:
            print("Already data is present")
        try:
           results = db_conn.execute(sqlalchemy.text("SELECT * FROM employeetable WHERE id=1")).fetchall()
        except:
            print("Select statement not working")
        
    return str(results)