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)] |