I am new to SQLAlchemy which is used to create db connections used in my org. As of now, logic to connect to DB is present which works on local systems. But when running automation from a different server, it does not work and I get below error :
<code>Failed to connect to database: (psycopg2.OperationalError) connection to server at "<aws-rds-server>" (<ip>), port <port> failed: Operation timed out
Is the server running on that host and accepting TCP/IP connections?
connection to server at "<aws-rds-server>" (<ip>), port <port> failed: Operation timed out
Is the server running on that host and accepting TCP/IP connections?
</code>
<code>Failed to connect to database: (psycopg2.OperationalError) connection to server at "<aws-rds-server>" (<ip>), port <port> failed: Operation timed out
Is the server running on that host and accepting TCP/IP connections?
connection to server at "<aws-rds-server>" (<ip>), port <port> failed: Operation timed out
Is the server running on that host and accepting TCP/IP connections?
</code>
Failed to connect to database: (psycopg2.OperationalError) connection to server at "<aws-rds-server>" (<ip>), port <port> failed: Operation timed out
Is the server running on that host and accepting TCP/IP connections?
connection to server at "<aws-rds-server>" (<ip>), port <port> failed: Operation timed out
Is the server running on that host and accepting TCP/IP connections?
I tried to run my code on completely different server and it works fine. I am not sure what to do next.
Below is the sample code of how I am connecting to Database:
<code>def connect_to_db(self) -> None:
"""Connect to Postgres database."""
sql_connection_url = URL.create(
drivername="postgresql",
username=self.db_username,
password=self.db_password,
database=self.db_name,
host=self.db_hostname,
port=self.db_port,
)
print(sql_connection_url)
try:
engine = create_engine(
sql_connection_url, echo=True
)
self.db_connection = engine.connect()
print(f"Connected to database: {self.db_hostname}:{self.db_port}/{self.db_name}")
except SQLAlchemyError as sql_error:
raise SQLAlchemyError(f"Failed to connect to database: {sql_error}") from sql_error
</code>
<code>def connect_to_db(self) -> None:
"""Connect to Postgres database."""
sql_connection_url = URL.create(
drivername="postgresql",
username=self.db_username,
password=self.db_password,
database=self.db_name,
host=self.db_hostname,
port=self.db_port,
)
print(sql_connection_url)
try:
engine = create_engine(
sql_connection_url, echo=True
)
self.db_connection = engine.connect()
print(f"Connected to database: {self.db_hostname}:{self.db_port}/{self.db_name}")
except SQLAlchemyError as sql_error:
raise SQLAlchemyError(f"Failed to connect to database: {sql_error}") from sql_error
</code>
def connect_to_db(self) -> None:
"""Connect to Postgres database."""
sql_connection_url = URL.create(
drivername="postgresql",
username=self.db_username,
password=self.db_password,
database=self.db_name,
host=self.db_hostname,
port=self.db_port,
)
print(sql_connection_url)
try:
engine = create_engine(
sql_connection_url, echo=True
)
self.db_connection = engine.connect()
print(f"Connected to database: {self.db_hostname}:{self.db_port}/{self.db_name}")
except SQLAlchemyError as sql_error:
raise SQLAlchemyError(f"Failed to connect to database: {sql_error}") from sql_error