matitos_news/utils/DB-Dev.ipynb

In [ ]:

#!pip install python-dotenv
from dotenv import load_dotenv

# Specify the path to your .env file (optional if in the current dir)
load_dotenv(dotenv_path=".env", override=True)

import os
import psycopg
from sshtunnel import SSHTunnelForwarder

if (os.environ.get("SSH_TUNNEL_BASED") == "true"):
    print("SSH tunnel: True")
else:
    print("SSH tunnel: False")

connect_info = "host={} port={} user={} password={} dbname={}".format(os.environ.get("DB_HOST"), os.environ.get("DB_PORT"), os.environ.get("DB_USER"), os.environ.get("DB_PASSWORD"), os.environ.get("DB_NAME"))

In [ ]:

if (os.environ.get("SSH_TUNNEL_BASED") == "true"):
    ssh_tunnel = SSHTunnelForwarder(
        (os.environ.get("REMOTE_HOST"), int(os.environ.get("REMOTE_SSH_PORT"))), 
        ssh_username=os.environ.get("REMOTE_USERNAME"), ssh_password=os.environ.get("REMOTE_PASSWORD"), 
        remote_bind_address=('localhost', int(os.environ.get("REMOTE_PORT"))), local_bind_address=('localhost', int(os.environ.get("DB_PORT"))) 
    )
    ssh_tunnel.start()

try:
    with psycopg.connect(connect_info) as conn:
        if True:
            for t in conn.execute("""
                        SELECT * from URLS WHERE id IN (SELECT id_url FROM URLS_SOURCE_SEARCH INNER JOIN SEARCH ON URLS_SOURCE_SEARCH.id_search = SEARCH.id WHERE SEARCH.search LIKE '%child abuse%') LIMIT 5;
                    """).fetchall():
                print(t)
            
except Exception as e:
    print("Err:", str(e))

if (os.environ.get("SSH_TUNNEL_BASED") == "true"):
    ssh_tunnel.stop()