Hologres:Python

Prerequisites

Python 3.7 or later is installed.

Install Psycopg 3

pip  install   --upgrade pip             # Upgrade pip to version 20.3 or later.
pip install "psycopg[binary]"

Connect to Hologres

Once Psycopg 3 is installed, you can connect to Hologres.

1. Import Psycopg 3.

   Import the Psycopg 3 library:

   import psycopg

2. Create a database connection.

   To connect to Hologres, use the psycopg.connect() function with the following parameters:

   conn = psycopg.connect(
       host="<endpoint>",
       port=<Port>, 
       dbname="<database_name>", 
       user="<AccessKey ID>", 
       password="<AccessKey Secret>",
       keepalives=<keepalives>, 
       keepalives_idle=<keepalives_idle>,
       keepalives_interval=<keepalives_interval>, 
       keepalives_count=<keepalives_count>
   )
   

   Parameter	Description
   endpoint	The endpoint and port of the Hologres instance. Go to the Hologres console. In the left-side navigation pane, click Instances and click the target instance. In the Network Information section of the Instance Details page, select the Network Type that matches the network environment where your code runs, and copy its Domain Name. Important Ensure you select the correct endpoint and port for your network environment to prevent connection failures.
   port
   dbname	The name of the database that you created in Hologres.
   AccessKey ID	The AccessKey ID of your Alibaba Cloud account. To obtain an AccessKey ID, go to the AccessKey page.
   AccessKey Secret	The AccessKey Secret of your Alibaba Cloud account.
   keepalives	Recommended. Specifies whether to use a persistent connection. Valid values: 1: uses a persistent connection. 0: uses a non-persistent connection.
   keepalives_idle	The idle time in seconds before a keepalive probe is sent.
   keepalives_interval	The time in seconds to wait for a response to a keepalive probe before retransmitting.
   keepalives_count	The maximum number of keepalive probes that can be sent before the connection is considered lost.

   The following code provides an example.

   conn = psycopg.connect(
       host="<endpoint>",
       port=<Port>, 
       dbname="<database_name>", 
       user="<AccessKey ID>", 
       password="<AccessKey Secret>",
       keepalives=1, # Enable persistent connection.
       keepalives_idle=130, # Send a keepalive probe every 130 seconds on an idle connection.
       keepalives_interval=10, # Wait 10 seconds for a response before retransmitting.
       keepalives_count=15, # Retransmit up to 15 times before closing the connection.
       application_name="<Application Name>"
   )

   Note

   Setting the application_name parameter helps you quickly identify the source application of a query in the historical slow query list.

Use Hologres

After connecting to your Hologres database, you can use Psycopg 3 to perform data operations, such as creating tables, inserting data, querying data, and releasing resources. To achieve higher read and write performance with the Fixed Plan feature, you must configure the related GUC parameters. For more information, see Accelerate SQL execution by using Fixed Plan.

1. Create a cursor.

   Before you perform data operations, you must run the cur = conn.cursor() command to create a cursor for the connection.

2. Perform data operations.

   1. Create a table

      Run the following command to create a table named holo_test with an integer column.

      cur.execute("CREATE TABLE holo_test (num integer);")

   2. Insert data

      Run the following command to insert the integers from 1 to 1000 into the holo_test table.

      cur.execute("INSERT INTO holo_test SELECT generate_series(%s, %s)", (1, 1000))

   3. Query data

      cur.execute("SELECT sum(num) FROM holo_test;")
      cur.fetchone()

3. Commit the transaction.

   By default, Psycopg starts a transaction that you must explicitly commit by using conn.commit(). For convenience, we recommend setting the autocommit parameter to True to commit each command automatically. The following examples show how to do this:

   • Synchronous call example

     conn = psycopg.connect(
         host="<endpoint>",
         port=<Port>, 
         dbname="<database_name>", 
         user="<AccessKey ID>", 
         password="<AccessKey Secret>",
         keepalives=1, # Enable persistent connection.
         keepalives_idle=130, # Send a keepalive probe every 130 seconds on an idle connection.
         keepalives_interval=10, # Wait 10 seconds for a response before retransmitting.
         keepalives_count=15, # Retransmit up to 15 times before closing the connection.
         application_name="<Application Name>"
     )
     conn.autocommit = True

   • Asynchronous call example

     async with await psycopg.AsyncConnection.connect(
         host="<endpoint>",
         port=<Port>, 
         dbname="<database_name>", 
         user="<AccessKey ID>", 
         password="<AccessKey Secret>",
         application_name="<Application Name>",
         autocommit = True
         ) as aconn:
         async with aconn.cursor() as acur:
             await acur.execute(
                 "INSERT INTO test (num, data) VALUES (%s, %s)",
                 (100, "abc'def"))
             await acur.execute("SELECT * FROM test")
             await acur.fetchone()
             # will return (1, 100, "abc'def")
             async for record in acur:
                 print(record)

4. Release resources.

   When you are finished, close the cursor and the database connection to release resources.

   cur.close()
   conn.close()

Best practice: Write a DataFrame to Hologres

A common workflow in Python is to process data in a Pandas DataFrame before importing it into Hologres. This section demonstrates an efficient method for this import.

# pip install Pandas==1.5.1

import psycopg
import pandas as pd

# Connect to Hologres.
conn = psycopg.connect(
    host="hgpostcn-cn-xxxxx-cn-hangzhou.hologres.aliyuncs.com",
    port=80,
    dbname="db",
    user="xxx",
    password="xxx",
    application_name="psycopg3"
)

cur = conn.cursor()

# Drop the table if it already exists.
cur.execute("""
            DROP TABLE IF EXISTS df_data;
            """)
conn.commit()

# Create a test table for data ingestion.
cur.execute("""
            CREATE TABLE IF NOT EXISTS df_data(
                col1 int,
                col2 int,
                col3 int,
                primary key(col1)
            );
            """)
conn.commit()

# Create a DataFrame.
data = [('1','1','1'),('2','2','2')]
cols = ('col1','col2','col3')
pd_data = pd.DataFrame(data, columns=cols)

# Use StringIO to convert the DataFrame into a CSV-formatted string.
from io import StringIO

# Create an in-memory buffer.
buffer = StringIO()
        
# Write the DataFrame to the buffer in CSV format.
pd_data.to_csv(buffer, index=False, header=False)
        
# Reset the buffer's position to the beginning.
buffer.seek(0)

with cur.copy("COPY df_data(col1,col2,col3) FROM STDIN WITH (STREAM_MODE TRUE,ON_CONFLICT UPDATE,FORMAT CSV);") as copy:
    while data := buffer.read(1024):
        copy.write(data)
conn.commit()

# Query the data.
cur.execute("SELECT * FROM df_data")
cur.fetchone()
cur.commit()