FilesExpand file tree

 main

/

example_basic.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

89 lines (70 loc) · 2.61 KB

 main

/

example_basic.py

Copy path

Top

File metadata and controls

• Code
• Blame

89 lines (70 loc) · 2.61 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

import ibm_db

import ibm_db_dbi

import pandas as pd

from testcontainers.db2 import Db2Container

def basic_example():

with Db2Container() as db2:

# Get connection parameters

host = db2.get_container_host_ip()

port = db2.get_exposed_port(db2.port)

database = db2.database

username = db2.username

password = db2.password

# Create connection string

conn_str = f"DATABASE={database};HOSTNAME={host};PORT={port};PROTOCOL=TCPIP;UID={username};PWD={password}"

# Connect to DB2

conn = ibm_db.connect(conn_str, "", "")

print("Connected to DB2")

# Create a test table

create_table_sql = """

CREATE TABLE test_table (

id INTEGER NOT NULL PRIMARY KEY,

name VARCHAR(50),

value DECIMAL(10,2),

created_at TIMESTAMP DEFAULT CURRENT TIMESTAMP

)

"""

try:

ibm_db.exec_immediate(conn, create_table_sql)

print("Created test table")

except Exception as e:

print(f"Table might already exist: {e}")

# Insert test data

test_data = [(1, "test1", 100.0), (2, "test2", 200.0), (3, "test3", 300.0)]

insert_sql = "INSERT INTO test_table (id, name, value) VALUES (?, ?, ?)"

stmt = ibm_db.prepare(conn, insert_sql)

for row in test_data:

ibm_db.execute(stmt, row)

print("Inserted test data")

# Query data using ibm_db_dbi

conn_dbi = ibm_db_dbi.Connection(conn)

cursor = conn_dbi.cursor()

cursor.execute("SELECT * FROM test_table ORDER BY id")

rows = cursor.fetchall()

print("\nQuery results:")

for row in rows:

print(f"ID: {row[0]}, Name: {row[1]}, Value: {row[2]}, Created: {row[3]}")

# Execute a more complex query

cursor.execute("""

SELECT

name,

AVG(value) as avg_value,

COUNT(*) as count,

MIN(created_at) as first_created,

MAX(created_at) as last_created

FROM test_table

GROUP BY name

ORDER BY avg_value DESC

""")

print("\nAggregation results:")

for row in cursor.fetchall():

print(f"Name: {row[0]}, Avg: {row[1]:.2f}, Count: {row[2]}, First: {row[3]}, Last: {row[4]}")

# Convert to pandas DataFrame

df = pd.read_sql("SELECT * FROM test_table ORDER BY id", conn_dbi)

print("\nDataFrame:")

print(df)

# Clean up

cursor.close()

ibm_db.close(conn)

if __name__ == "__main__":

basic_example()