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 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 | import pytest
from pkg_resources import parse_version
from sqlalchemy import Column
from sqlalchemy import Float
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import __version__ as SA_VERSION
from sqlalchemy.ext.declarative import declarative_base
from geoalchemy2 import Raster
from geoalchemy2 import WKTElement
from geoalchemy2.functions import GenericFunction
from geoalchemy2.types import CompositeType
# Tests imports
from tests import select
from tests import test_only_with_dialects
class SummaryStatsCustomType(CompositeType):
"""Define the composite type returned by the function ST_SummaryStatsAgg."""
typemap = {
'count': Integer,
'sum': Float,
'mean': Float,
'stddev': Float,
'min': Float,
'max': Float,
}
cache_ok = True
class ST_SummaryStatsAgg(GenericFunction):
type = SummaryStatsCustomType
# Set a specific identifier to not override the actual ST_SummaryStatsAgg function
identifier = "ST_SummaryStatsAgg_custom"
inherit_cache = True
metadata = MetaData()
Base = declarative_base(metadata=metadata)
class Ocean(Base):
__tablename__ = 'ocean'
id = Column(Integer, primary_key=True)
rast = Column(Raster)
def __init__(self, rast):
self.rast = rast
@test_only_with_dialects("postgresql")
class TestSTSummaryStatsAgg():
@pytest.mark.skipif(
parse_version(SA_VERSION) < parse_version("1.4"),
reason="requires SQLAlchely>1.4",
)
def test_st_summary_stats_agg(self, session, conn):
metadata.drop_all(conn, checkfirst=True)
metadata.create_all(conn)
# Create a new raster
polygon = WKTElement('POLYGON((0 0,1 1,0 1,0 0))', srid=4326)
o = Ocean(polygon.ST_AsRaster(5, 6))
session.add(o)
session.flush()
# Define the query to compute stats
stats_agg = select([
Ocean.rast.ST_SummaryStatsAgg_custom(1, True, 1).label("stats")
])
stats_agg_alias = stats_agg.alias("stats_agg")
# Use these stats
query = select([
stats_agg_alias.c.stats.count.label("count"),
stats_agg_alias.c.stats.sum.label("sum"),
stats_agg_alias.c.stats.mean.label("mean"),
stats_agg_alias.c.stats.stddev.label("stddev"),
stats_agg_alias.c.stats.min.label("min"),
stats_agg_alias.c.stats.max.label("max")
])
# Check the query
assert str(query.compile(dialect=session.bind.dialect)) == (
"SELECT "
"(stats_agg.stats).count AS count, "
"(stats_agg.stats).sum AS sum, "
"(stats_agg.stats).mean AS mean, "
"(stats_agg.stats).stddev AS stddev, "
"(stats_agg.stats).min AS min, "
"(stats_agg.stats).max AS max \n"
"FROM ("
"SELECT "
"ST_SummaryStatsAgg("
"ocean.rast, "
"%(ST_SummaryStatsAgg_1)s, %(ST_SummaryStatsAgg_2)s, %(ST_SummaryStatsAgg_3)s"
") AS stats \n"
"FROM ocean) AS stats_agg"
)
# Execute the query
res = session.execute(query).fetchall()
# Check the result
assert res == [(15, 15.0, 1.0, 0.0, 1.0, 1.0)]
|
