forked from testcontainers/testcontainers-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathexample_basic.py
More file actions
161 lines (137 loc) · 4.79 KB
/
example_basic.py
File metadata and controls
161 lines (137 loc) · 4.79 KB
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
import pymssql
from testcontainers.mssql import MsSqlContainer
def basic_example():
with MsSqlContainer() as mssql:
# Get connection parameters
host = mssql.get_container_host_ip()
port = mssql.get_exposed_port(mssql.port)
username = mssql.username
password = mssql.password
database = mssql.database
# Connect to MSSQL
connection = pymssql.connect(server=host, port=port, user=username, password=password, database=database)
print("Connected to MSSQL")
# Create cursor
cursor = connection.cursor()
# Create test table
cursor.execute("""
CREATE TABLE test_table (
id INT IDENTITY(1,1) PRIMARY KEY,
name NVARCHAR(50),
value INT,
category NVARCHAR(10),
created_at DATETIME2 DEFAULT GETDATE()
)
""")
print("Created test table")
# Insert test data
test_data = [("test1", 100, "A"), ("test2", 200, "B"), ("test3", 300, "A")]
cursor.executemany(
"""
INSERT INTO test_table (name, value, category)
VALUES (%s, %s, %s)
""",
test_data,
)
print("Inserted test data")
# Commit changes
connection.commit()
# Query data
print("\nQuery results:")
cursor.execute("SELECT * FROM test_table WHERE category = 'A'")
for row in cursor:
print({"id": row[0], "name": row[1], "value": row[2], "category": row[3], "created_at": row[4].isoformat()})
# Create view
cursor.execute("""
CREATE OR ALTER VIEW test_view AS
SELECT category, COUNT(*) as count, AVG(value) as avg_value
FROM test_table
GROUP BY category
""")
print("\nCreated view")
# Query view
print("\nView results:")
cursor.execute("SELECT * FROM test_view")
for row in cursor:
print({"category": row[0], "count": row[1], "avg_value": float(row[2])})
# Create index
cursor.execute("CREATE INDEX test_idx ON test_table (value)")
print("\nCreated index")
# Query using index
print("\nQuery using index:")
cursor.execute("SELECT * FROM test_table WHERE value > 150")
for row in cursor:
print({"id": row[0], "name": row[1], "value": row[2], "category": row[3], "created_at": row[4].isoformat()})
# Get table metadata
cursor.execute("""
SELECT
c.name as column_name,
t.name as data_type,
c.max_length,
c.is_nullable
FROM sys.columns c
JOIN sys.types t ON c.user_type_id = t.user_type_id
WHERE OBJECT_ID = OBJECT_ID('test_table')
ORDER BY c.column_id
""")
print("\nTable metadata:")
for row in cursor:
print({"column": row[0], "type": row[1], "length": row[2], "nullable": row[3]})
# Create stored procedure
cursor.execute("""
CREATE OR ALTER PROCEDURE test_proc
@category NVARCHAR(10),
@count INT OUTPUT
AS
BEGIN
SELECT @count = COUNT(*)
FROM test_table
WHERE category = @category
END
""")
print("\nCreated stored procedure")
# Execute stored procedure
cursor.execute("""
DECLARE @count INT
EXEC test_proc @category = 'A', @count = @count OUTPUT
SELECT @count as count
""")
count = cursor.fetchone()[0]
print(f"Count for category A: {count}")
# Create function
cursor.execute("""
CREATE OR ALTER FUNCTION test_func(@category NVARCHAR(10))
RETURNS TABLE
AS
RETURN
(
SELECT name, value
FROM test_table
WHERE category = @category
)
""")
print("\nCreated function")
# Use function
print("\nFunction results:")
cursor.execute("SELECT * FROM test_func('A')")
for row in cursor:
print({"name": row[0], "value": row[1]})
# Create trigger
cursor.execute("""
CREATE OR ALTER TRIGGER test_trigger
ON test_table
AFTER INSERT
AS
BEGIN
PRINT 'New row inserted'
END
""")
print("\nCreated trigger")
# Test trigger
cursor.execute("INSERT INTO test_table (name, value, category) VALUES ('test4', 400, 'B')")
connection.commit()
# Clean up
cursor.close()
connection.close()
if __name__ == "__main__":
basic_example()