Note
Click here to download the full example code
Use CompositeTypeΒΆ
Some functions return composite types. This example shows how to deal with this kind of functions.
8 import pytest
9 from pkg_resources import parse_version
10
11 from sqlalchemy import __version__ as SA_VERSION
12 from sqlalchemy import Column
13 from sqlalchemy import create_engine
14 from sqlalchemy import Float
15 from sqlalchemy import Integer
16 from sqlalchemy import MetaData
17 from sqlalchemy import select
18 from sqlalchemy.ext.declarative import declarative_base
19 from sqlalchemy.orm import sessionmaker
20
21 from geoalchemy2 import Raster, WKTElement
22 from geoalchemy2.functions import GenericFunction
23 from geoalchemy2.types import CompositeType
24
25
26 class SummaryStatsCustomType(CompositeType):
27 """Define the composite type returned by the function ST_SummaryStatsAgg."""
28 typemap = {
29 'count': Integer,
30 'sum': Float,
31 'mean': Float,
32 'stddev': Float,
33 'min': Float,
34 'max': Float,
35 }
36
37 cache_ok = True
38
39
40 class ST_SummaryStatsAgg(GenericFunction):
41 type = SummaryStatsCustomType
42 # Set a specific identifier to not override the actual ST_SummaryStatsAgg function
43 identifier = "ST_SummaryStatsAgg_custom"
44
45 inherit_cache = True
46
47
48 engine = create_engine('postgresql://gis:gis@localhost/gis', echo=True)
49 metadata = MetaData(engine)
50 Base = declarative_base(metadata=metadata)
51 session = sessionmaker(bind=engine)()
52
53
54 class Ocean(Base):
55 __tablename__ = 'ocean'
56 __table_args__ = {'schema': 'public'}
57 id = Column(Integer, primary_key=True)
58 rast = Column(Raster)
59
60 def __init__(self, rast):
61 self.rast = rast
62
63
64 class TestSTSummaryStatsAgg():
65
66 def setup(self):
67 metadata.drop_all(checkfirst=True)
68 metadata.create_all()
69
70 def teardown(self):
71 session.rollback()
72 metadata.drop_all()
73
74 @pytest.mark.skipif(
75 parse_version(SA_VERSION) < parse_version("1.4"),
76 reason="requires SQLAlchely>1.4",
77 )
78 def test_st_summary_stats_agg(self):
79
80 # Create a new raster
81 polygon = WKTElement('POLYGON((0 0,1 1,0 1,0 0))', srid=4326)
82 o = Ocean(polygon.ST_AsRaster(5, 6))
83 session.add(o)
84 session.flush()
85
86 # Define the query to compute stats
87 stats_agg = select(
88 Ocean.rast.ST_SummaryStatsAgg_custom(1, True, 1).label("stats")
89 )
90 stats_agg_alias = stats_agg.alias("stats_agg")
91
92 # Use these stats
93 query = select(
94 stats_agg_alias.c.stats.count.label("count"),
95 stats_agg_alias.c.stats.sum.label("sum"),
96 stats_agg_alias.c.stats.mean.label("mean"),
97 stats_agg_alias.c.stats.stddev.label("stddev"),
98 stats_agg_alias.c.stats.min.label("min"),
99 stats_agg_alias.c.stats.max.label("max")
100 )
101
102 # Check the query
103 assert str(query) == (
104 "SELECT "
105 "(stats_agg.stats).count AS count, "
106 "(stats_agg.stats).sum AS sum, "
107 "(stats_agg.stats).mean AS mean, "
108 "(stats_agg.stats).stddev AS stddev, "
109 "(stats_agg.stats).min AS min, "
110 "(stats_agg.stats).max AS max \n"
111 "FROM ("
112 "SELECT "
113 "ST_SummaryStatsAgg("
114 "public.ocean.rast, "
115 "%(ST_SummaryStatsAgg_1)s, %(ST_SummaryStatsAgg_2)s, %(ST_SummaryStatsAgg_3)s"
116 ") AS stats \n"
117 "FROM public.ocean) AS stats_agg"
118 )
119
120 # Execute the query
121 res = session.execute(query).fetchall()
122
123 # Check the result
124 assert res == [(15, 15.0, 1.0, 0.0, 1.0, 1.0)]
Total running time of the script: ( 0 minutes 0.000 seconds)