FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
testcontainers-python/modules/postgres/example_basic.py at main · DhavalGojiya/testcontainers-python · GitHub
DhavalGojiya
/
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
0
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
/
postgres
/
example_basic.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
99 lines (81 loc) · 3.06 KB
Breadcrumbs
testcontainers-python
/
modules
/
postgres
/
example_basic.py
Copy path
File metadata and controls
99 lines (81 loc) · 3.06 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
91
92
93
94
95
96
97
98
99
import
pandas
as
pd
import
sqlalchemy
from
sqlalchemy
import
text
from
testcontainers
.
postgres
import
PostgresContainer
def
basic_example
():
with
PostgresContainer
()
as
postgres
:
# Get connection URL
connection_url
=
postgres
.
get_connection_url
()
# Create SQLAlchemy engine
engine
=
sqlalchemy
.
create_engine
(
connection_url
)
print
(
"Connected to PostgreSQL"
)
# Create a test table
create_table_sql
=
"""
CREATE TABLE test_table (
id SERIAL PRIMARY KEY,
name VARCHAR(50),
value DECIMAL(10,2),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
with
engine
.
begin
()
as
connection
:
connection
.
execute
(
text
(
create_table_sql
))
print
(
"Created test table"
)
# Insert test data
test_data
=
[
{
"name"
:
"test1"
,
"value"
:
100.0
},
{
"name"
:
"test2"
,
"value"
:
200.0
},
{
"name"
:
"test3"
,
"value"
:
300.0
},
]
with
engine
.
begin
()
as
connection
:
for
data
in
test_data
:
connection
.
execute
(
text
(
"INSERT INTO test_table (name, value) VALUES (:name, :value)"
),
data
)
print
(
"Inserted test data"
)
# Query data
with
engine
.
connect
()
as
connection
:
result
=
connection
.
execute
(
text
(
"SELECT * FROM test_table ORDER BY id"
))
rows
=
result
.
fetchall
()
print
(
"
\n
Query results:"
)
for
row
in
rows
:
print
(
f"ID:
{
row
[
0
]
}
, Name:
{
row
[
1
]
}
, Value:
{
row
[
2
]
}
, Created:
{
row
[
3
]
}
"
)
# Execute a more complex query
with
engine
.
connect
()
as
connection
:
result
=
connection
.
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
[
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"
,
engine
)
print
(
"
\n
DataFrame:"
)
print
(
df
)
# Create and query a view
create_view_sql
=
"""
CREATE OR REPLACE VIEW test_view AS
SELECT
name,
AVG(value) as avg_value,
COUNT(*) as count
FROM test_table
GROUP BY name
"""
with
engine
.
begin
()
as
connection
:
connection
.
execute
(
text
(
create_view_sql
))
print
(
"
\n
Created view"
)
result
=
connection
.
execute
(
text
(
"SELECT * FROM test_view"
))
print
(
"
\n
View results:"
)
for
row
in
result
:
print
(
f"Name:
{
row
[
0
]
}
, Avg:
{
row
[
1
]:.2f
}
, Count:
{
row
[
2
]
}
"
)
if
__name__
==
"__main__"
:
basic_example
()
Back
|
FazBrowse Home
|
New Git URL