Can't connect to MySQL database with Paramiko, can with sshtunnel

My code was using sshtunnel and connected to my database correctly. However due to a conflict between grequests and sshtunnel I had to switch to using another library for the ssh, and have had most luck with Paramiko (as that works with monkeypatch on grequests).

The ssh tunnel section is working correctly, however I am now getting the message:

An error occurred: 1045 (28000): Access denied for user 'PA_username'@'localhost' (using password: YES)

I'm using environment variables that haven't changed, as far as I can see everything is the same between the working code with sshtunnel and without grequests, and the version with Paramiko that can't connect.

Full code here:

from gevent import monkey

monkey.patch_all()

import grequests
import mysql.connector
import paramiko
from pprint import pprint
import os

ssh_password = os.environ.get("PA_SSH_PASS")
db_password = os.environ.get("PA_DB_PASS")
database = os.environ.get("PA_DB_NAME")

def get_current_listing_urls():
    try:
        ssh_client = paramiko.SSHClient()
        ssh_client.load_system_host_keys()
        ssh_client.set_missing_host_key_policy(paramiko.WarningPolicy)

        ssh_client.connect(
            "ssh.eu.pythonanywhere.com",
            username=PA_username,
            password=ssh_password,
        )

        transport = ssh_client.get_transport()

        channel = transport.open_channel(
            "direct-tcpip",
            dest_addr=("PA_username.mysql.eu.pythonanywhere-services.com", 3306),
            src_addr=("127.0.0.1", 3306),
        )

        db = mysql.connector.connect(
            user="PA_username",
            password=db_password,
            host="127.0.0.1",
            port=3306,
            database=database,
            use_pure=True,
        )

        table_name = "listings"

        cursor = db.cursor(dictionary=True)

        query = f"SELECT COUNT(*) FROM {table_name};"

        cursor.execute(query)

        results = cursor.fetchall()

        return results

    except Exception as e:
        print(f"An error occurred: {str(e)}")
    finally:
        if "cursor" in locals() and cursor:
            cursor.close()
        if "db" in locals() and db:
            db.close()
        if "ssh_client" in locals() and ssh_client:
            ssh_client.close()


old_listing_urls = get_current_listing_urls()

pprint(old_listing_urls)

I've tried just about everything I can think of, but I can't see where I'm going wrong. The message presumably means it's connecting and trying to authenticate, but failing. However the database name, username, and password are all correct as they are all working when using sshtunnel.

Any help would be much appreciated! Thanks

EDIT:

The working code with sshtunnel is here:

import mysql.connector
import sshtunnel
from pprint import pprint
import time
import os
import json

sshtunnel.SSH_TIMEOUT = 5.0
sshtunnel.TUNNEL_TIMEOUT = 5.0

ssh_password = os.environ.get("PA_SSH_PASS")
db_password = os.environ.get("PA_DB_PASS")
database = os.environ.get("PA_DB_NAME")


def get_current_listing_urls():
    try:
        with sshtunnel.SSHTunnelForwarder(
            ("ssh.eu.pythonanywhere.com"),
            ssh_username=PA_username,
            ssh_password=ssh_password,
            remote_bind_address=(
                "PA_username.mysql.eu.pythonanywhere-services.com",
                3306,
            ),
            threaded=False,
        ) as tunnel:
            db = mysql.connector.connect(
                user=PA_username,
                password=db_password,
                host="127.0.0.1",
                port=tunnel.local_bind_port,
                database=database,
                use_pure=True,
            )

            table_name = "listings"

            cursor = db.cursor(dictionary=True)

            query = f"SELECT COUNT(*)  FROM {table_name};"  

            cursor.execute(query)

            results = cursor.fetchall()

            return results

    except Exception as e:
        print(f"An error occurred: {str(e)}")
    finally:
        if "cursor" in locals() and cursor:
            cursor.close()
        if "db" in locals() and db:
            db.close()
        if "tunnel" in locals() and tunnel:
            tunnel.close()


old_listing_urls = get_current_listing_urls()

pprint(old_listing_urls)

I wasn't able to solve this how I wanted to, in the end I used subprocess to run an ssh command in cmd and close it afterwards. This needed an rsa key instead of a password as I couldn't enter the password into cmd after sending the ssh command. Normally pevent would be used for this, but the function needed (spawn) doesn't work on windows. Using a key means no password entry is required.

Full code:

import subprocess
from db_utilities import get_current_listing_urls

# Define the SSH command to establish a tunnel
ssh_command = "ssh -L 3333:PA_username.mysql.pythonanywhere-services.com:3306 PA_username@ssh.pythonanywhere.com"

# Open a new cmd window and run the SSH command
cmd_process = subprocess.Popen(
    ["cmd", "/K", ssh_command],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    universal_newlines=True,
    text=True,
)

try:
    results = get_current_listing_urls()
    print(results)
finally:
    # Check if the process is still running, and terminate it if necessary
    if cmd_process.poll() is None:
        print("Closing SSH Tunnel")
        cmd_process.terminate()