FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
testcontainers-python/modules/cockroachdb/example_basic.py at main · f18m/testcontainers-python · GitHub
f18m
/
testcontainers-python
Public
forked from
testcontainers/testcontainers-python
Notifications
You must be signed in to change notification settings
Fork
0
Star
0
Code
Pull requests
1
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
testcontainers-python
/
modules
/
cockroachdb
/
example_basic.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
90 lines (76 loc) · 2.75 KB
Breadcrumbs
testcontainers-python
/
modules
/
cockroachdb
/
example_basic.py
Copy path
File metadata and controls
90 lines (76 loc) · 2.75 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
90
import
pandas
as
pd
import
sqlalchemy
from
sqlalchemy
import
text
from
testcontainers
.
cockroachdb
import
CockroachContainer
def
basic_example
():
with
CockroachContainer
()
as
cockroach
:
# Get connection URL
connection_url
=
cockroach
.
get_connection_url
()
# Create SQLAlchemy engine
engine
=
sqlalchemy
.
create_engine
(
connection_url
)
# Create a test table
with
engine
.
begin
()
as
conn
:
conn
.
execute
(
text
(
"""
CREATE TABLE IF NOT EXISTS test_table (
id SERIAL PRIMARY KEY,
name VARCHAR(50),
value DECIMAL(10,2),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
print
(
"Created test table"
)
# Insert test data
test_data
=
[(
1
,
"test1"
,
100.0
), (
2
,
"test2"
,
200.0
), (
3
,
"test3"
,
300.0
)]
conn
.
execute
(
text
(
"""
INSERT INTO test_table (id, name, value)
VALUES (:id, :name, :value)
"""
),
[{
"id"
:
item_id
,
"name"
:
name
,
"value"
:
value
}
for
item_id
,
name
,
value
in
test_data
],
)
print
(
"Inserted test data"
)
# Query data
with
engine
.
connect
()
as
conn
:
result
=
conn
.
execute
(
text
(
"""
SELECT *
FROM test_table
ORDER BY id
"""
)
)
print
(
"
\n
Query results:"
)
for
row
in
result
:
print
(
f"ID:
{
row
.
id
}
, Name:
{
row
.
name
}
, Value:
{
row
.
value
}
, Created:
{
row
.
created_at
}
"
)
# Execute a more complex query
with
engine
.
connect
()
as
conn
:
result
=
conn
.
execute
(
text
(
"""
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
(
"
\n
Aggregation results:"
)
for
row
in
result
:
print
(
f"Name:
{
row
.
name
}
, "
f"Avg:
{
row
.
avg_value
:.2f
}
, "
f"Count:
{
row
.
count
}
, "
f"First:
{
row
.
first_created
}
, "
f"Last:
{
row
.
last_created
}
"
)
# Convert to pandas DataFrame
with
engine
.
connect
()
as
conn
:
df
=
pd
.
read_sql
(
"SELECT * FROM test_table ORDER BY id"
,
conn
)
print
(
"
\n
DataFrame:"
)
print
(
df
)
if
__name__
==
"__main__"
:
basic_example
()
Back
|
FazBrowse Home
|
New Git URL