Skip to content

📘 API Reference Features

Georama Features

admin

ColumnOgcApiFeaturesInline

Bases: TabularInline

Source code in src/georama/features/admin.py 
18
19
20
21
22
23
24
25
class ColumnOgcApiFeaturesInline(admin.TabularInline):
    model = ColumnOgcApiFeatures
    formset = ColumnOgcApiFeaturesInlineFormset
    can_delete = False
    extra = 0

    def has_add_permission(self, request, obj):
        return False

can_delete = False class-attribute instance-attribute

extra = 0 class-attribute instance-attribute

formset = ColumnOgcApiFeaturesInlineFormset class-attribute instance-attribute

model = ColumnOgcApiFeatures class-attribute instance-attribute

has_add_permission(request, obj)

Source code in src/georama/features/admin.py 
24
25
def has_add_permission(self, request, obj):
    return False

ColumnOgcApiFeaturesInlineFormset

Bases: BaseInlineFormSet

Source code in src/georama/features/admin.py 
13
14
15
class ColumnOgcApiFeaturesInlineFormset(BaseInlineFormSet):
    model = ColumnOgcApiFeatures
    fields = ["name", "title"]

fields = ['name', 'title'] class-attribute instance-attribute

model = ColumnOgcApiFeatures class-attribute instance-attribute

PublishedAsOgcApiFeaturesAdmin

Bases: ModelAdmin

Source code in src/georama/features/admin.py 
 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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@admin.register(PublishedAsOgcApiFeatures)
class PublishedAsOgcApiFeaturesAdmin(admin.ModelAdmin):
    list_display = ["name", "title", "public", "delete_link", "show_published"]
    inlines = [ColumnOgcApiFeaturesInline]
    add_form_template = "admin/features/publishedasvectorfeature/publish.html"
    list_editable = ["public"]
    readonly_fields = ["dataset"]

    form = PublishedAsOgcApiFeaturesForm

    fieldsets = (
        (
            None,
            {
                "fields": (
                    "title",
                    "name",
                    "public",
                    "column_permission",
                    "description",
                    "license",
                    "fees",
                    "access_constraints",
                    "dataset",
                    "max_items",
                    "default_items",
                    "on_exceed",
                )
            },
        ),
        (
            "Group permissions",
            {
                "fields": (
                    "group_read_permission",
                    "group_create_permission",
                    "group_update_permission",
                    "group_delete_permission",
                )
            },
        ),
        (
            "User permissions",
            {
                "fields": (
                    "user_read_permission",
                    "user_create_permission",
                    "user_update_permission",
                    "user_delete_permission",
                )
            },
        ),
    )

    def add_view(self, request, form_url="", extra_context=None):
        extra_context = extra_context or {}
        vector_datasets = VectorDataSet.objects.all()
        extra_context["vector_datasets"] = [
            (vd, reverse("publish_as_oapif", args=[vd.id])) for vd in vector_datasets
        ]
        return super().add_view(
            request,
            form_url,
            extra_context=extra_context,
        )

    def delete_link(self, obj: PublishedAsOgcApiFeatures):
        return mark_safe(
            '<a href="{}" class="btn btn-high btn-success">&#128465;</a>'.format(
                reverse("admin:features_publishedasogcapifeatures_delete", args=(obj.pk,))
            )
        )

    def show_published(self, obj: PublishedAsOgcApiFeatures):
        return mark_safe(
            '<a href="{}" class="btn btn-high btn-success">&#128065;</a>'.format(
                reverse("collection-detail", args=(str(obj.identifier),))
            )
        )

    def save_model(self, request, obj, form, change):
        permissions_dct = {"read": "", "create": "", "update": "", "delete": ""}

        # query the correct permission object
        for permission_type in permissions_dct.keys():
            permissions_dct[permission_type] = Permission.objects.get(
                codename=_get_permissions(obj, permission_type)
            )

        # save the permissions
        for permission_type, permission in permissions_dct.items():
            # save group permissions
            groups = form.cleaned_data.get(f"group_{permission_type}_permission", [])
            save_group_permissions(groups, permission)
            # save user permissions
            users = form.cleaned_data.get(f"user_{permission_type}_permission", [])
            save_user_permissions(users, permission)

        super().save_model(request, obj, form, change)

add_form_template = 'admin/features/publishedasvectorfeature/publish.html' class-attribute instance-attribute

fieldsets = ((None, {'fields': ('title', 'name', 'public', 'column_permission', 'description', 'license', 'fees', 'access_constraints', 'dataset', 'max_items', 'default_items', 'on_exceed')}), ('Group permissions', {'fields': ('group_read_permission', 'group_create_permission', 'group_update_permission', 'group_delete_permission')}), ('User permissions', {'fields': ('user_read_permission', 'user_create_permission', 'user_update_permission', 'user_delete_permission')})) class-attribute instance-attribute

form = PublishedAsOgcApiFeaturesForm class-attribute instance-attribute

inlines = [ColumnOgcApiFeaturesInline] class-attribute instance-attribute

list_display = ['name', 'title', 'public', 'delete_link', 'show_published'] class-attribute instance-attribute

list_editable = ['public'] class-attribute instance-attribute

readonly_fields = ['dataset'] class-attribute instance-attribute

add_view(request, form_url='', extra_context=None)

Source code in src/georama/features/admin.py 
87
88
89
90
91
92
93
94
95
96
97
def add_view(self, request, form_url="", extra_context=None):
    extra_context = extra_context or {}
    vector_datasets = VectorDataSet.objects.all()
    extra_context["vector_datasets"] = [
        (vd, reverse("publish_as_oapif", args=[vd.id])) for vd in vector_datasets
    ]
    return super().add_view(
        request,
        form_url,
        extra_context=extra_context,
    )
Source code in src/georama/features/admin.py 
 99
100
101
102
103
104
def delete_link(self, obj: PublishedAsOgcApiFeatures):
    return mark_safe(
        '<a href="{}" class="btn btn-high btn-success">&#128465;</a>'.format(
            reverse("admin:features_publishedasogcapifeatures_delete", args=(obj.pk,))
        )
    )

save_model(request, obj, form, change)

Source code in src/georama/features/admin.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def save_model(self, request, obj, form, change):
    permissions_dct = {"read": "", "create": "", "update": "", "delete": ""}

    # query the correct permission object
    for permission_type in permissions_dct.keys():
        permissions_dct[permission_type] = Permission.objects.get(
            codename=_get_permissions(obj, permission_type)
        )

    # save the permissions
    for permission_type, permission in permissions_dct.items():
        # save group permissions
        groups = form.cleaned_data.get(f"group_{permission_type}_permission", [])
        save_group_permissions(groups, permission)
        # save user permissions
        users = form.cleaned_data.get(f"user_{permission_type}_permission", [])
        save_user_permissions(users, permission)

    super().save_model(request, obj, form, change)

show_published(obj)

Source code in src/georama/features/admin.py 
106
107
108
109
110
111
def show_published(self, obj: PublishedAsOgcApiFeatures):
    return mark_safe(
        '<a href="{}" class="btn btn-high btn-success">&#128065;</a>'.format(
            reverse("collection-detail", args=(str(obj.identifier),))
        )
    )

_get_permissions(obj, permission_type)

Source code in src/georama/features/admin.py 
28
29
30
def _get_permissions(obj: PublishedAsOgcApiFeatures, permission_type: str):
    permissions = obj.permissions
    return [p.codename for p in permissions if permission_type in p.codename][0]

apps

appname = 'features' module-attribute

VectorparrotConfig

Bases: AppConfig

Source code in src/georama/features/apps.py 
20
21
22
class VectorparrotConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = f"georama.{appname}"

default_auto_field = 'django.db.models.BigAutoField' class-attribute instance-attribute

name = f'georama.{appname}' class-attribute instance-attribute

config_openapi

config_openapi()

Source code in src/georama/features/config_openapi.py 
  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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def config_openapi() -> dict:
    return {
        "components": {
            "parameters": {
                "bbox": {
                    "description": "Only features that have a geometry that intersects the bounding box are selected.The bounding box is provided as four or six numbers, depending on whether the coordinate reference system includes a vertical axis (height or depth).",
                    "explode": False,
                    "in": "query",
                    "name": "bbox",
                    "required": False,
                    "schema": {
                        "items": {"type": "number"},
                        "maxItems": 6,
                        "minItems": 4,
                        "type": "array",
                    },
                    "style": "form",
                },
                "bbox-crs": {
                    "description": "Indicates the coordinate reference system for the given bbox coordinates.",
                    "explode": False,
                    "in": "query",
                    "name": "bbox-crs",
                    "required": False,
                    "schema": {"format": "uri", "type": "string"},
                    "style": "form",
                },
                "bbox-crs-epsg": {
                    "description": "Indicates the EPSG for the given bbox coordinates.",
                    "explode": False,
                    "in": "query",
                    "name": "bbox-crs",
                    "required": False,
                    "schema": {"default": 4326, "type": "integer"},
                    "style": "form",
                },
                "crs": {
                    "description": "Indicates the coordinate reference system for the results.",
                    "explode": False,
                    "in": "query",
                    "name": "crs",
                    "required": False,
                    "schema": {"format": "uri", "type": "string"},
                    "style": "form",
                },
                "f": {
                    "description": "The optional f parameter indicates the output format which the server shall provide as part of the response document.  The default format is GeoJSON.",
                    "explode": False,
                    "in": "query",
                    "name": "f",
                    "required": False,
                    "schema": {
                        "default": "json",
                        "enum": ["json", "html", "jsonld"],
                        "type": "string",
                    },
                    "style": "form",
                },
                "lang": {
                    "description": 'The optional lang parameter instructs the server return a response in a certain language, if supported.  If the language is not among the available values, the Accept-Language header language will be used if it is supported. If the header is missing, the default server language is used. Note that providers may only support a single language (or often no language at all), that can be different from the server language.  Language strings can be written in a complex (e.g. "fr-CA,fr;q=0.9,en-US;q=0.8,en;q=0.7"), simple (e.g. "de") or locale-like (e.g. "de-CH" or "fr_BE") fashion.',
                    "in": "query",
                    "name": "lang",
                    "required": False,
                    "schema": {
                        "default": "en-US",
                        "enum": ["en-US", "fr-CA"],
                        "type": "string",
                    },
                },
                "offset": {
                    "description": "The optional offset parameter indicates the index within the result set from which the server shall begin presenting results in the response document.  The first element has an index of 0 (default).",
                    "explode": False,
                    "in": "query",
                    "name": "offset",
                    "required": False,
                    "schema": {"default": 0, "minimum": 0, "type": "integer"},
                    "style": "form",
                },
                "resourceId": {
                    "description": "Configuration resource identifier",
                    "in": "path",
                    "name": "resourceId",
                    "required": True,
                    "schema": {"type": "string"},
                },
                "skipGeometry": {
                    "description": "This option can be used to skip response geometries for each feature.",
                    "explode": False,
                    "in": "query",
                    "name": "skipGeometry",
                    "required": False,
                    "schema": {"default": False, "type": "boolean"},
                    "style": "form",
                },
                "vendorSpecificParameters": {
                    "description": 'Additional "free-form" parameters that are not explicitly defined',
                    "in": "query",
                    "name": "vendorSpecificParameters",
                    "schema": {"additionalProperties": True, "type": "object"},
                    "style": "form",
                },
            },
            "responses": {
                "200": {"description": "successful operation"},
                "204": {"description": "no content"},
                "Queryables": {
                    "content": {
                        "application/json": {
                            "schema": {"$ref": "#/components/schemas/queryables"}
                        }
                    },
                    "description": "successful queryables operation",
                },
                "Tiles": {
                    "content": {
                        "application/json": {"schema": {"$ref": "#/components/schemas/tiles"}}
                    },
                    "description": "Retrieves the tiles description for this collection",
                },
                "default": {
                    "content": {
                        "application/json": {
                            "schema": {
                                "$ref": "https://schemas.opengis.net/ogcapi/processes/part1/1.0/openapi/schemas/exception.yaml"
                            }
                        }
                    },
                    "description": "Unexpected error",
                },
            },
            "schemas": {
                "queryable": {
                    "properties": {
                        "description": {
                            "description": "a human-readable narrative describing the queryable",
                            "type": "string",
                        },
                        "language": {
                            "default": ["en"],
                            "description": "the language used for the title and description",
                            "type": "string",
                        },
                        "queryable": {
                            "description": "the token that may be used in a CQL predicate",
                            "type": "string",
                        },
                        "title": {
                            "description": "a human readable title for the queryable",
                            "type": "string",
                        },
                        "type": {
                            "description": "the data type of the queryable",
                            "type": "string",
                        },
                        "type-ref": {
                            "description": "a reference to the formal definition of the type",
                            "format": "url",
                            "type": "string",
                        },
                    },
                    "required": ["queryable", "type"],
                    "type": "object",
                },
                "queryables": {
                    "properties": {
                        "queryables": {
                            "items": {"$ref": "#/components/schemas/queryable"},
                            "type": "array",
                        }
                    },
                    "required": ["queryables"],
                    "type": "object",
                },
                "tilematrixsetlink": {
                    "properties": {
                        "tileMatrixSet": {"type": "string"},
                        "tileMatrixSetURI": {"type": "string"},
                    },
                    "required": ["tileMatrixSet"],
                    "type": "object",
                },
                "tiles": {
                    "properties": {
                        "links": {
                            "items": {
                                "$ref": "https://schemas.opengis.net/ogcapi/tiles/part1/1.0/openapi/ogcapi-tiles-1.yaml#/components/schemas/link"
                            },
                            "type": "array",
                        },
                        "tileMatrixSetLinks": {
                            "items": {"$ref": "#/components/schemas/tilematrixsetlink"},
                            "type": "array",
                        },
                    },
                    "required": ["tileMatrixSetLinks", "links"],
                    "type": "object",
                },
            },
        },
        "info": {
            "contact": {
                "email": "you@example.org",
                "name": "Organization Name",
                "url": "https://pygeoapi.io",
            },
            "description": "pygeoapi provides an API to geospatial data",
            "license": {
                "name": "CC-BY 4.0 license",
                "url": "https://creativecommons.org/licenses/by/4.0/",
            },
            "termsOfService": "None",
            "title": "pygeoapi default instance",
            "version": "3.0.2",
            "x-keywords": ["geospatial", "data", "api"],
        },
        "openapi": "3.0.2",
        "paths": {
            "/openapi": {
                "get": {
                    "description": "This document",
                    "operationId": "getOpenapi",
                    "parameters": [
                        {"$ref": "#/components/parameters/f"},
                        {"$ref": "#/components/parameters/lang"},
                        {
                            "description": "UI to render the OpenAPI document",
                            "explode": False,
                            "in": "query",
                            "name": "ui",
                            "required": False,
                            "schema": {
                                "default": "swagger",
                                "enum": ["swagger", "redoc"],
                                "type": "string",
                            },
                            "style": "form",
                        },
                    ],
                    "responses": {
                        "200": {"$ref": "#/components/responses/200"},
                        "400": {
                            "$ref": "https://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/ogcapi-features-1.yaml#/components/responses/InvalidParameter"
                        },
                        "default": {"$ref": "#/components/responses/default"},
                    },
                    "summary": "This document",
                    "tags": ["server"],
                }
            },
            "/collections": {
                "get": {
                    "description": "Feature Collections",
                    "operationId": "getCollections",
                    "parameters": [
                        {"$ref": "#/components/parameters/f"},
                        {"$ref": "#/components/parameters/lang"},
                    ],
                    "responses": {
                        "200": {
                            "$ref": "http://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/ogcapi-features-1.yaml#/components/responses/Collections"
                        },
                        "400": {
                            "$ref": "http://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/ogcapi-features-1.yaml#/components/responses/InvalidParameter"
                        },
                        "500": {
                            "$ref": "http://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/ogcapi-features-1.yaml#/components/responses/ServerError"
                        },
                    },
                    "summary": "Feature Collections",
                    "tags": ["server"],
                }
            },
            "/conformance": {
                "get": {
                    "description": "API conformance definition",
                    "operationId": "getConformanceDeclaration",
                    "parameters": [
                        {"$ref": "#/components/parameters/f"},
                        {"$ref": "#/components/parameters/lang"},
                    ],
                    "responses": {
                        "200": {
                            "$ref": "http://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/ogcapi-features-1.yaml#/components/responses/ConformanceDeclaration"
                        },
                        "400": {
                            "$ref": "http://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/ogcapi-features-1.yaml#/components/responses/InvalidParameter"
                        },
                        "500": {
                            "$ref": "http://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/ogcapi-features-1.yaml#/components/responses/ServerError"
                        },
                    },
                    "summary": "API conformance definition",
                    "tags": ["server"],
                }
            },
        },
        "servers": [
            {
                "description": "pygeoapi provides an API to geospatial data",
                "url": "OVERWRITTEN ON RUNTIME",
            }
        ],
        "tags": [
            {
                "description": "pygeoapi provides an API to geospatial data",
                "externalDocs": {"description": "information", "url": "http://example.org"},
                "name": "server",
            },
            {"description": "Kantonsgrenzen", "name": "kantonsgrenzen"},
        ],
    }

config_server

ServerConfig

Source code in src/georama/features/config_server.py 
 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
class ServerConfig:
    def get(self) -> dict:
        on_exceed_key, on_exceed_value = PublishedAsVectorFeature.ON_EXCEED_CHOICES[0]
        return {
            "server": {
                "url": "OVERWRITTEN AT RUNTIME",
                "mimetype": "application/json; charset=UTF-8",
                "encoding": "utf-8",
                "gzip": False,
                "languages": ["en-US", "fr-CA"],
                "pretty_print": True,
                "limits": {
                    "default_items": 10,
                    "max_items": 500,
                    "on_exceed": on_exceed_value,
                },
                "map": {
                    "url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
                    "attribution": '&copy; <a href="https://openstreetmap.org/copyright">OpenStreetMap contributors</a>',
                },
            },
            "logging": {"level": "INFO"},
            "metadata": {
                "identification": {
                    "title": {
                        "en": "pygeoapi default instance",
                        "fr": "instance par défaut de pygeoapi",
                    },
                    "description": {
                        "en": "pygeoapi provides an API to geospatial data",
                        "fr": "pygeoapi fournit une API aux données géospatiales",
                    },
                    "keywords": {
                        "en": ["geospatial", "data", "api"],
                        "fr": ["géospatiale", "données", "api"],
                    },
                    "keywords_type": "theme",
                    "terms_of_service": "https://creativecommons.org/licenses/by/4.0/",
                    "url": "https://example.org",
                },
                "license": {
                    "name": "CC-BY 4.0 license",
                    "url": "https://creativecommons.org/licenses/by/4.0/",
                },
                "provider": {"name": "Organization Name", "url": "https://opengis.ch"},
                "contact": {
                    "name": "Clemens, Rudert",
                    "position": "Scruffy",
                    "address": "clemens@opengis.ch",
                    "city": "Basel",
                    "stateorprovince": "Basel-Stadt",
                    "postalcode": "4058",
                    "country": "Switzerland",
                    "phone": "+xx-xxx-xxx-xxxx",
                    "fax": "+xx-xxx-xxx-xxxx",
                    "email": "clemens@opengis.ch",
                    "url": "https://opengis.ch",
                    "hours": "Mo-Fr 08:00-17:00",
                    "instructions": "During hours of service. Off on weekends.",
                    "role": "pointOfContact",
                },
            },
            "resources": {},
        }

get()

Source code in src/georama/features/config_server.py 
 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
def get(self) -> dict:
    on_exceed_key, on_exceed_value = PublishedAsVectorFeature.ON_EXCEED_CHOICES[0]
    return {
        "server": {
            "url": "OVERWRITTEN AT RUNTIME",
            "mimetype": "application/json; charset=UTF-8",
            "encoding": "utf-8",
            "gzip": False,
            "languages": ["en-US", "fr-CA"],
            "pretty_print": True,
            "limits": {
                "default_items": 10,
                "max_items": 500,
                "on_exceed": on_exceed_value,
            },
            "map": {
                "url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
                "attribution": '&copy; <a href="https://openstreetmap.org/copyright">OpenStreetMap contributors</a>',
            },
        },
        "logging": {"level": "INFO"},
        "metadata": {
            "identification": {
                "title": {
                    "en": "pygeoapi default instance",
                    "fr": "instance par défaut de pygeoapi",
                },
                "description": {
                    "en": "pygeoapi provides an API to geospatial data",
                    "fr": "pygeoapi fournit une API aux données géospatiales",
                },
                "keywords": {
                    "en": ["geospatial", "data", "api"],
                    "fr": ["géospatiale", "données", "api"],
                },
                "keywords_type": "theme",
                "terms_of_service": "https://creativecommons.org/licenses/by/4.0/",
                "url": "https://example.org",
            },
            "license": {
                "name": "CC-BY 4.0 license",
                "url": "https://creativecommons.org/licenses/by/4.0/",
            },
            "provider": {"name": "Organization Name", "url": "https://opengis.ch"},
            "contact": {
                "name": "Clemens, Rudert",
                "position": "Scruffy",
                "address": "clemens@opengis.ch",
                "city": "Basel",
                "stateorprovince": "Basel-Stadt",
                "postalcode": "4058",
                "country": "Switzerland",
                "phone": "+xx-xxx-xxx-xxxx",
                "fax": "+xx-xxx-xxx-xxxx",
                "email": "clemens@opengis.ch",
                "url": "https://opengis.ch",
                "hours": "Mo-Fr 08:00-17:00",
                "instructions": "During hours of service. Off on weekends.",
                "role": "pointOfContact",
            },
        },
        "resources": {},
    }

features_config

Config

Source code in src/georama/features/features_config.py 
 4
 5
 6
 7
 8
 9
10
11
class Config:
    @property
    def path(self) -> str:
        return os.path.join(os.environ.get("GEORAMA_DATA_INTEGRATION_ROOT", "/io/data"))

    @property
    def default_crs(self):
        return "https://www.opengis.net/def/crs/EPSG/0/2056"

default_crs property

path property

forms

PublishedAsOgcApiFeaturesForm

Bases: ModelForm

Source code in src/georama/features/forms.py 
 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
class PublishedAsOgcApiFeaturesForm(forms.ModelForm):
    class Meta:
        model = PublishedAsOgcApiFeatures
        fields = "__all__"

    group_read_permission = forms.ModelMultipleChoiceField(
        required=False,
        queryset=None,
        widget=widgets.FilteredSelectMultiple(
            verbose_name="Group read permission", is_stacked=False
        ),
    )
    group_create_permission = forms.ModelMultipleChoiceField(
        required=False,
        queryset=None,
        widget=widgets.FilteredSelectMultiple(
            verbose_name="Group create permission", is_stacked=False
        ),
    )
    group_update_permission = forms.ModelMultipleChoiceField(
        required=False,
        queryset=None,
        widget=widgets.FilteredSelectMultiple(
            verbose_name="Group update permission", is_stacked=False
        ),
    )
    group_delete_permission = forms.ModelMultipleChoiceField(
        required=False,
        queryset=None,
        widget=widgets.FilteredSelectMultiple(
            verbose_name="Group delete permission", is_stacked=False
        ),
    )
    user_read_permission = forms.ModelMultipleChoiceField(
        required=False,
        queryset=None,
        widget=widgets.FilteredSelectMultiple(
            verbose_name="User read permission", is_stacked=False
        ),
    )
    user_create_permission = forms.ModelMultipleChoiceField(
        required=False,
        queryset=None,
        widget=widgets.FilteredSelectMultiple(
            verbose_name="User create permission", is_stacked=False
        ),
    )
    user_update_permission = forms.ModelMultipleChoiceField(
        required=False,
        queryset=None,
        widget=widgets.FilteredSelectMultiple(
            verbose_name="User update permission", is_stacked=False
        ),
    )
    user_delete_permission = forms.ModelMultipleChoiceField(
        required=False,
        queryset=None,
        widget=widgets.FilteredSelectMultiple(
            verbose_name="User delete permission", is_stacked=False
        ),
    )

    @staticmethod
    def get_first_match_partial_key(dictionary, partial_key) -> List[str]:
        matches = [k for k, v in dictionary.items() if partial_key in k]
        return matches if matches else None

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # a dict to map the fields to the permissions
        perm_fields = {field: "" for field in self.fields if "permission" in field}
        if kwargs.get("instance"):

            # getting the permissions of the object
            permissions = kwargs["instance"].permissions
            permission_codenames = [p.codename for p in permissions]

            # mapping the permission to the field by part of the dict key
            # f.e. field "group_read_permission" to 'wms_read_bdb158db-3501-4563-be37-b2369ccf64e6'
            for perm in permission_codenames:
                permission_keys = PublishedAsOgcApiFeaturesForm.get_first_match_partial_key(
                    perm_fields, perm.split("_")[1]
                )
                if len(permission_keys) > 0:
                    for p_key in permission_keys:
                        perm_fields[p_key] = perm

            # filling the field with the correct queryset and initialize it with the data from the db
            for field, perm in perm_fields.items():
                if field == "column_permission":
                    pass
                elif perm.strip() != "":
                    if "group" in field:
                        self.fields[field].queryset = Group.objects.all()
                        self.fields[field].initial = Group.objects.filter(
                            permissions__codename=perm
                        )
                    elif "user" in field:
                        self.fields[field].queryset = User.objects.all().exclude(
                            is_superuser=True
                        )
                        self.fields[field].initial = User.objects.filter(
                            user_permissions__codename=perm
                        ).exclude(is_superuser=True)
                    else:
                        raise Exception(f"Unknown permission field: {field}")

group_create_permission = forms.ModelMultipleChoiceField(required=False, queryset=None, widget=widgets.FilteredSelectMultiple(verbose_name='Group create permission', is_stacked=False)) class-attribute instance-attribute

group_delete_permission = forms.ModelMultipleChoiceField(required=False, queryset=None, widget=widgets.FilteredSelectMultiple(verbose_name='Group delete permission', is_stacked=False)) class-attribute instance-attribute

group_read_permission = forms.ModelMultipleChoiceField(required=False, queryset=None, widget=widgets.FilteredSelectMultiple(verbose_name='Group read permission', is_stacked=False)) class-attribute instance-attribute

group_update_permission = forms.ModelMultipleChoiceField(required=False, queryset=None, widget=widgets.FilteredSelectMultiple(verbose_name='Group update permission', is_stacked=False)) class-attribute instance-attribute

user_create_permission = forms.ModelMultipleChoiceField(required=False, queryset=None, widget=widgets.FilteredSelectMultiple(verbose_name='User create permission', is_stacked=False)) class-attribute instance-attribute

user_delete_permission = forms.ModelMultipleChoiceField(required=False, queryset=None, widget=widgets.FilteredSelectMultiple(verbose_name='User delete permission', is_stacked=False)) class-attribute instance-attribute

user_read_permission = forms.ModelMultipleChoiceField(required=False, queryset=None, widget=widgets.FilteredSelectMultiple(verbose_name='User read permission', is_stacked=False)) class-attribute instance-attribute

user_update_permission = forms.ModelMultipleChoiceField(required=False, queryset=None, widget=widgets.FilteredSelectMultiple(verbose_name='User update permission', is_stacked=False)) class-attribute instance-attribute

Meta

Source code in src/georama/features/forms.py 
11
12
13
class Meta:
    model = PublishedAsOgcApiFeatures
    fields = "__all__"
fields = '__all__' class-attribute instance-attribute
model = PublishedAsOgcApiFeatures class-attribute instance-attribute

__init__(*args, **kwargs)

Source code in src/georama/features/forms.py 
 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
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)

    # a dict to map the fields to the permissions
    perm_fields = {field: "" for field in self.fields if "permission" in field}
    if kwargs.get("instance"):

        # getting the permissions of the object
        permissions = kwargs["instance"].permissions
        permission_codenames = [p.codename for p in permissions]

        # mapping the permission to the field by part of the dict key
        # f.e. field "group_read_permission" to 'wms_read_bdb158db-3501-4563-be37-b2369ccf64e6'
        for perm in permission_codenames:
            permission_keys = PublishedAsOgcApiFeaturesForm.get_first_match_partial_key(
                perm_fields, perm.split("_")[1]
            )
            if len(permission_keys) > 0:
                for p_key in permission_keys:
                    perm_fields[p_key] = perm

        # filling the field with the correct queryset and initialize it with the data from the db
        for field, perm in perm_fields.items():
            if field == "column_permission":
                pass
            elif perm.strip() != "":
                if "group" in field:
                    self.fields[field].queryset = Group.objects.all()
                    self.fields[field].initial = Group.objects.filter(
                        permissions__codename=perm
                    )
                elif "user" in field:
                    self.fields[field].queryset = User.objects.all().exclude(
                        is_superuser=True
                    )
                    self.fields[field].initial = User.objects.filter(
                        user_permissions__codename=perm
                    ).exclude(is_superuser=True)
                else:
                    raise Exception(f"Unknown permission field: {field}")

get_first_match_partial_key(dictionary, partial_key) staticmethod

Source code in src/georama/features/forms.py 
72
73
74
75
@staticmethod
def get_first_match_partial_key(dictionary, partial_key) -> List[str]:
    matches = [k for k, v in dictionary.items() if partial_key in k]
    return matches if matches else None

migrations

0001_initial

Migration

Bases: Migration

Source code in src/georama/features/migrations/0001_initial.py 
  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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class Migration(migrations.Migration):

    initial = True

    dependencies = [
        ("data_integration", "0020_remove_vectordataset_extent_buffer"),
    ]

    operations = [
        migrations.CreateModel(
            name="PublishedAsOgcApiFeatures",
            fields=[
                (
                    "identifier",
                    models.UUIDField(
                        default=uuid.uuid4, editable=False, primary_key=True, serialize=False
                    ),
                ),
                (
                    "name",
                    models.CharField(blank=True, default=None, max_length=1000, null=True),
                ),
                ("public", models.BooleanField(default=False)),
                (
                    "title",
                    models.CharField(blank=True, default=None, max_length=1000, null=True),
                ),
                ("description", models.TextField(blank=True, default=None, null=True)),
                (
                    "license",
                    models.TextField(
                        default="\n    This dataset is made available under the Open Database\n    License: http://opendatacommons.org/licenses/odbl/1.0/.\n    Any rights in individual contents of the database are licensed\n    under the Database Contents\n    License: http://opendatacommons.org/licenses/dbcl/1.0/\n    "
                    ),
                ),
                ("fees", models.TextField(default="No fees apply.")),
                (
                    "access_constraints",
                    models.TextField(default="No access constraints apply."),
                ),
                ("column_permission", models.BooleanField(default=False)),
                (
                    "dataset",
                    models.ForeignKey(
                        null=True,
                        on_delete=django.db.models.deletion.CASCADE,
                        related_name="published_ogc_api_features",
                        related_query_name="published_ogc_api_feature",
                        to="data_integration.vectordataset",
                    ),
                ),
            ],
            options={
                "abstract": False,
            },
        ),
        migrations.CreateModel(
            name="ColumnOgcApiFeatures",
            fields=[
                (
                    "identifier",
                    models.UUIDField(
                        default=uuid.uuid4, editable=False, primary_key=True, serialize=False
                    ),
                ),
                (
                    "name",
                    models.CharField(blank=True, default=None, max_length=1000, null=True),
                ),
                ("public", models.BooleanField(default=False)),
                ("title", models.CharField(max_length=1000)),
                (
                    "published_definition",
                    models.ForeignKey(
                        on_delete=django.db.models.deletion.CASCADE,
                        related_name="columns",
                        related_query_name="column",
                        to="features.publishedasogcapifeatures",
                    ),
                ),
            ],
            options={
                "abstract": False,
            },
        ),
        migrations.CreateModel(
            name="PublishedAsWfs",
            fields=[
                (
                    "identifier",
                    models.UUIDField(
                        default=uuid.uuid4, editable=False, primary_key=True, serialize=False
                    ),
                ),
                (
                    "name",
                    models.CharField(blank=True, default=None, max_length=1000, null=True),
                ),
                ("public", models.BooleanField(default=False)),
                (
                    "title",
                    models.CharField(blank=True, default=None, max_length=1000, null=True),
                ),
                ("description", models.TextField(blank=True, default=None, null=True)),
                (
                    "license",
                    models.TextField(
                        default="\n    This dataset is made available under the Open Database\n    License: http://opendatacommons.org/licenses/odbl/1.0/.\n    Any rights in individual contents of the database are licensed\n    under the Database Contents\n    License: http://opendatacommons.org/licenses/dbcl/1.0/\n    "
                    ),
                ),
                ("fees", models.TextField(default="No fees apply.")),
                (
                    "access_constraints",
                    models.TextField(default="No access constraints apply."),
                ),
                ("column_permission", models.BooleanField(default=False)),
                (
                    "dataset",
                    models.ForeignKey(
                        null=True,
                        on_delete=django.db.models.deletion.CASCADE,
                        related_name="published_ogc_wfs",
                        related_query_name="published_ogc_wfs",
                        to="data_integration.vectordataset",
                    ),
                ),
            ],
            options={
                "abstract": False,
            },
        ),
        migrations.CreateModel(
            name="ColumnWfs",
            fields=[
                (
                    "identifier",
                    models.UUIDField(
                        default=uuid.uuid4, editable=False, primary_key=True, serialize=False
                    ),
                ),
                (
                    "name",
                    models.CharField(blank=True, default=None, max_length=1000, null=True),
                ),
                ("public", models.BooleanField(default=False)),
                ("title", models.CharField(max_length=1000)),
                (
                    "published_definition",
                    models.ForeignKey(
                        on_delete=django.db.models.deletion.CASCADE,
                        related_name="columns",
                        related_query_name="column",
                        to="features.publishedaswfs",
                    ),
                ),
            ],
            options={
                "abstract": False,
            },
        ),
    ]
dependencies = [('data_integration', '0020_remove_vectordataset_extent_buffer')] class-attribute instance-attribute
initial = True class-attribute instance-attribute
operations = [migrations.CreateModel(name='PublishedAsOgcApiFeatures', fields=[('identifier', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), ('name', models.CharField(blank=True, default=None, max_length=1000, null=True)), ('public', models.BooleanField(default=False)), ('title', models.CharField(blank=True, default=None, max_length=1000, null=True)), ('description', models.TextField(blank=True, default=None, null=True)), ('license', models.TextField(default='\n This dataset is made available under the Open Database\n License: http://opendatacommons.org/licenses/odbl/1.0/.\n Any rights in individual contents of the database are licensed\n under the Database Contents\n License: http://opendatacommons.org/licenses/dbcl/1.0/\n ')), ('fees', models.TextField(default='No fees apply.')), ('access_constraints', models.TextField(default='No access constraints apply.')), ('column_permission', models.BooleanField(default=False)), ('dataset', models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='published_ogc_api_features', related_query_name='published_ogc_api_feature', to='data_integration.vectordataset'))], options={'abstract': False}), migrations.CreateModel(name='ColumnOgcApiFeatures', fields=[('identifier', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), ('name', models.CharField(blank=True, default=None, max_length=1000, null=True)), ('public', models.BooleanField(default=False)), ('title', models.CharField(max_length=1000)), ('published_definition', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='columns', related_query_name='column', to='features.publishedasogcapifeatures'))], options={'abstract': False}), migrations.CreateModel(name='PublishedAsWfs', fields=[('identifier', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), ('name', models.CharField(blank=True, default=None, max_length=1000, null=True)), ('public', models.BooleanField(default=False)), ('title', models.CharField(blank=True, default=None, max_length=1000, null=True)), ('description', models.TextField(blank=True, default=None, null=True)), ('license', models.TextField(default='\n This dataset is made available under the Open Database\n License: http://opendatacommons.org/licenses/odbl/1.0/.\n Any rights in individual contents of the database are licensed\n under the Database Contents\n License: http://opendatacommons.org/licenses/dbcl/1.0/\n ')), ('fees', models.TextField(default='No fees apply.')), ('access_constraints', models.TextField(default='No access constraints apply.')), ('column_permission', models.BooleanField(default=False)), ('dataset', models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='published_ogc_wfs', related_query_name='published_ogc_wfs', to='data_integration.vectordataset'))], options={'abstract': False}), migrations.CreateModel(name='ColumnWfs', fields=[('identifier', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), ('name', models.CharField(blank=True, default=None, max_length=1000, null=True)), ('public', models.BooleanField(default=False)), ('title', models.CharField(max_length=1000)), ('published_definition', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='columns', related_query_name='column', to='features.publishedaswfs'))], options={'abstract': False})] class-attribute instance-attribute

0002_publishedasogcapifeatures_default_items_and_more

Migration

Bases: Migration

Source code in src/georama/features/migrations/0002_publishedasogcapifeatures_default_items_and_more.py 
 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
class Migration(migrations.Migration):

    dependencies = [
        ("features", "0001_initial"),
    ]

    operations = [
        migrations.AddField(
            model_name="publishedasogcapifeatures",
            name="default_items",
            field=models.IntegerField(default=10, null=True),
        ),
        migrations.AddField(
            model_name="publishedasogcapifeatures",
            name="max_items",
            field=models.IntegerField(default=500, null=True),
        ),
        migrations.AddField(
            model_name="publishedasogcapifeatures",
            name="on_exceed",
            field=models.CharField(
                choices=[("ERROR", "error"), ("THROTTLE", "throttle")],
                default="ERROR",
                max_length=10,
                null=True,
            ),
        ),
        migrations.AddField(
            model_name="publishedaswfs",
            name="default_items",
            field=models.IntegerField(default=10, null=True),
        ),
        migrations.AddField(
            model_name="publishedaswfs",
            name="max_items",
            field=models.IntegerField(default=500, null=True),
        ),
        migrations.AddField(
            model_name="publishedaswfs",
            name="on_exceed",
            field=models.CharField(
                choices=[("ERROR", "error"), ("THROTTLE", "throttle")],
                default="ERROR",
                max_length=10,
                null=True,
            ),
        ),
    ]
dependencies = [('features', '0001_initial')] class-attribute instance-attribute
operations = [migrations.AddField(model_name='publishedasogcapifeatures', name='default_items', field=models.IntegerField(default=10, null=True)), migrations.AddField(model_name='publishedasogcapifeatures', name='max_items', field=models.IntegerField(default=500, null=True)), migrations.AddField(model_name='publishedasogcapifeatures', name='on_exceed', field=models.CharField(choices=[('ERROR', 'error'), ('THROTTLE', 'throttle')], default='ERROR', max_length=10, null=True)), migrations.AddField(model_name='publishedaswfs', name='default_items', field=models.IntegerField(default=10, null=True)), migrations.AddField(model_name='publishedaswfs', name='max_items', field=models.IntegerField(default=500, null=True)), migrations.AddField(model_name='publishedaswfs', name='on_exceed', field=models.CharField(choices=[('ERROR', 'error'), ('THROTTLE', 'throttle')], default='ERROR', max_length=10, null=True))] class-attribute instance-attribute

models

COLUMN_TYPE_VALUES = {'numeric': 'Numerical Type', 'string': 'Alphanumerical Type', 'boolean': 'Boolean Type', 'date': 'Date Type', 'datetime': 'Datetime Type'} module-attribute

Column

Bases: PublishedAsRoleNameSystem

Source code in src/georama/features/models.py 
 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
class Column(PublishedAsRoleNameSystem):
    published_as_type = "feature_column"
    title = models.CharField(max_length=1000)

    @classmethod
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        models.signals.pre_delete.connect(delete_publishedas_db_permissions, sender=cls)

    @property
    def create_permissions(self) -> List[PermissionInterface]:
        """delete permission not relevant for specific property: create/delete property happens at the layer level"""
        return []

    @property
    def delete_permissions(self) -> List[PermissionInterface]:
        """delete permission not relevant for specific property: create/delete property happens at the layer level"""
        return []

    def get_published_definition(self) -> PublishedAsVectorFeature:
        raise NotImplementedError

    def save(self, *args, **kwargs):
        if self.name is None:
            self.name = f"{self.get_published_definition().name}.{self.title}"
        super().save(*args, **kwargs)
        save_publishedas_db_permissions(self)

    @property
    def readable_identifier(self) -> str:
        """Using the publicatoin identifier at the end to make column visibly linked to their publication"""
        return f"{self.get_published_definition().readable_identifier}.{self.name}"

    class Meta:
        abstract = True

create_permissions property

delete permission not relevant for specific property: create/delete property happens at the layer level

delete_permissions property

delete permission not relevant for specific property: create/delete property happens at the layer level

published_as_type = 'feature_column' class-attribute instance-attribute

readable_identifier property

Using the publicatoin identifier at the end to make column visibly linked to their publication

title = models.CharField(max_length=1000) class-attribute instance-attribute

Meta

Source code in src/georama/features/models.py 
102
103
class Meta:
    abstract = True
abstract = True class-attribute instance-attribute

__init_subclass__(**kwargs) classmethod

Source code in src/georama/features/models.py 
73
74
75
76
@classmethod
def __init_subclass__(cls, **kwargs):
    super().__init_subclass__(**kwargs)
    models.signals.pre_delete.connect(delete_publishedas_db_permissions, sender=cls)

get_published_definition()

Source code in src/georama/features/models.py 
88
89
def get_published_definition(self) -> PublishedAsVectorFeature:
    raise NotImplementedError

save(*args, **kwargs)

Source code in src/georama/features/models.py 
91
92
93
94
95
def save(self, *args, **kwargs):
    if self.name is None:
        self.name = f"{self.get_published_definition().name}.{self.title}"
    super().save(*args, **kwargs)
    save_publishedas_db_permissions(self)

ColumnOgcApiFeatures

Bases: Column

Source code in src/georama/features/models.py 
178
179
180
181
182
183
184
185
186
187
class ColumnOgcApiFeatures(Column):
    published_definition = models.ForeignKey(
        PublishedAsOgcApiFeatures,
        related_name="columns",
        related_query_name="column",
        on_delete=models.CASCADE,
    )

    def get_published_definition(self) -> PublishedAsOgcApiFeatures:
        return self.published_definition

published_definition = models.ForeignKey(PublishedAsOgcApiFeatures, related_name='columns', related_query_name='column', on_delete=models.CASCADE) class-attribute instance-attribute

get_published_definition()

Source code in src/georama/features/models.py 
186
187
def get_published_definition(self) -> PublishedAsOgcApiFeatures:
    return self.published_definition

ColumnWfs

Bases: Column

Source code in src/georama/features/models.py 
122
123
124
125
126
127
128
129
130
131
class ColumnWfs(Column):
    published_definition = models.ForeignKey(
        PublishedAsWfs,
        related_name="columns",
        related_query_name="column",
        on_delete=models.CASCADE,
    )

    def get_published_definition(self) -> PublishedAsWfs:
        return self.published_definition

published_definition = models.ForeignKey(PublishedAsWfs, related_name='columns', related_query_name='column', on_delete=models.CASCADE) class-attribute instance-attribute

get_published_definition()

Source code in src/georama/features/models.py 
130
131
def get_published_definition(self) -> PublishedAsWfs:
    return self.published_definition

PublishedAsOgcApiFeatures

Bases: PublishedAsVectorFeature

Source code in src/georama/features/models.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
class PublishedAsOgcApiFeatures(PublishedAsVectorFeature):
    dataset = models.ForeignKey(
        VectorDataSet,
        # TODO: this seems wrong => only because error:
        #  It is impossible to add a non-nullable field 'dataset' to ... without
        #  specifying a default. This is because the database needs something to populate existing rows.
        null=True,
        related_name="published_ogc_api_features",
        related_query_name="published_ogc_api_feature",
        on_delete=models.CASCADE,
    )

    @property
    def readable_identifier(self) -> str:
        dataset = self.dataset
        return f"{dataset.project.mandant.name}.{dataset.project.name}.{dataset.name}.{self.identifier}"

    def get_columns(self) -> List["ColumnOgcApiFeatures"]:
        return self.columns.all()

    def save(self, force_insert=False, force_update=False, using=None, update_fields=None):
        if self.name is None and isinstance(self.dataset, VectorDataSet):
            # TODO: maybe we want this to be configurable?
            self.name = f"{self.dataset.project.mandant.name}.{self.dataset.project.name}.{self.dataset.name}"
        if self.title is None and isinstance(self.dataset, VectorDataSet):
            self.title = self.dataset.title
        super().save(
            force_insert=force_insert,
            force_update=force_update,
            using=using,
            update_fields=update_fields,
        )
        for field in self.dataset.fields.all():
            if not ColumnOgcApiFeatures.objects.filter(
                name=field.name, published_definition=self
            ).exists():
                ColumnOgcApiFeatures(
                    published_definition=self,
                    name=field.name,
                    title=field.name.title(),
                    public=True,
                ).save()

dataset = models.ForeignKey(VectorDataSet, null=True, related_name='published_ogc_api_features', related_query_name='published_ogc_api_feature', on_delete=models.CASCADE) class-attribute instance-attribute

readable_identifier property

get_columns()

Source code in src/georama/features/models.py 
151
152
def get_columns(self) -> List["ColumnOgcApiFeatures"]:
    return self.columns.all()

save(force_insert=False, force_update=False, using=None, update_fields=None)

Source code in src/georama/features/models.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def save(self, force_insert=False, force_update=False, using=None, update_fields=None):
    if self.name is None and isinstance(self.dataset, VectorDataSet):
        # TODO: maybe we want this to be configurable?
        self.name = f"{self.dataset.project.mandant.name}.{self.dataset.project.name}.{self.dataset.name}"
    if self.title is None and isinstance(self.dataset, VectorDataSet):
        self.title = self.dataset.title
    super().save(
        force_insert=force_insert,
        force_update=force_update,
        using=using,
        update_fields=update_fields,
    )
    for field in self.dataset.fields.all():
        if not ColumnOgcApiFeatures.objects.filter(
            name=field.name, published_definition=self
        ).exists():
            ColumnOgcApiFeatures(
                published_definition=self,
                name=field.name,
                title=field.name.title(),
                public=True,
            ).save()

PublishedAsVectorFeature

Bases: PublishedAs

Source code in src/georama/features/models.py 
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
class PublishedAsVectorFeature(PublishedAs):
    ON_EXCEED_CHOICES = (
        ("ERROR", "error"),
        ("THROTTLE", "throttle"),
    )
    published_as_type = "feature"
    column_permission = models.BooleanField(default=False)
    default_items = models.IntegerField(default=10, null=True)
    max_items = models.IntegerField(default=500, null=True)
    on_exceed = models.CharField(
        default="ERROR", choices=ON_EXCEED_CHOICES, max_length=10, null=True
    )

    class Meta:
        abstract = True

    def get_columns(self) -> List["Column"]:
        raise NotImplementedError

    @property
    def columns_permissions(self) -> List[PermissionInterface]:
        """Returns all the possible permissions for columns of this VectorFeature

        Doesn't check if column permissions are enabled on this VectorFeature publication"""
        return [p for col in self.get_columns() for p in col.permissions]

    @property
    def all_permissions(self) -> List[PermissionInterface]:
        permissions = self.permissions
        if self.column_permission:
            permissions = permissions + self.columns_permissions
        return permissions

    def has_general_permission(self, user: User, app_name: str) -> bool:
        """include columns permissions in this check"""
        if self.public:
            return True
        permissions = (
            self.permission_codenames
            if not self.column_permission
            else [p.codename for p in self.all_permissions]
        )
        return self._has_grained_permission(user, permissions, app_name)

ON_EXCEED_CHOICES = (('ERROR', 'error'), ('THROTTLE', 'throttle')) class-attribute instance-attribute

all_permissions property

column_permission = models.BooleanField(default=False) class-attribute instance-attribute

columns_permissions property

Returns all the possible permissions for columns of this VectorFeature

Doesn't check if column permissions are enabled on this VectorFeature publication

default_items = models.IntegerField(default=10, null=True) class-attribute instance-attribute

max_items = models.IntegerField(default=500, null=True) class-attribute instance-attribute

on_exceed = models.CharField(default='ERROR', choices=ON_EXCEED_CHOICES, max_length=10, null=True) class-attribute instance-attribute

published_as_type = 'feature' class-attribute instance-attribute

Meta

Source code in src/georama/features/models.py 
37
38
class Meta:
    abstract = True
abstract = True class-attribute instance-attribute

get_columns()

Source code in src/georama/features/models.py 
40
41
def get_columns(self) -> List["Column"]:
    raise NotImplementedError

has_general_permission(user, app_name)

include columns permissions in this check

Source code in src/georama/features/models.py 
57
58
59
60
61
62
63
64
65
66
def has_general_permission(self, user: User, app_name: str) -> bool:
    """include columns permissions in this check"""
    if self.public:
        return True
    permissions = (
        self.permission_codenames
        if not self.column_permission
        else [p.codename for p in self.all_permissions]
    )
    return self._has_grained_permission(user, permissions, app_name)

PublishedAsWfs

Bases: PublishedAsVectorFeature

Source code in src/georama/features/models.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class PublishedAsWfs(PublishedAsVectorFeature):
    dataset = models.ForeignKey(
        VectorDataSet,
        # TODO: this seems wrong => only because error:
        #  It is impossible to add a non-nullable field 'dataset' to ... without
        #  specifying a default. This is because the database needs something to populate existing rows.
        null=True,
        related_name="published_ogc_wfs",
        related_query_name="published_ogc_wfs",
        on_delete=models.CASCADE,
    )

    def get_columns(self) -> List["ColumnWfs"]:
        return self.columns.all()

dataset = models.ForeignKey(VectorDataSet, null=True, related_name='published_ogc_wfs', related_query_name='published_ogc_wfs', on_delete=models.CASCADE) class-attribute instance-attribute

get_columns()

Source code in src/georama/features/models.py 
118
119
def get_columns(self) -> List["ColumnWfs"]:
    return self.columns.all()

pygeoapi_providers

postgres

LOGGER = logging.getLogger(__name__) module-attribute

PostgresProvider

Bases: BaseProvider

Generic provider for Postgresql based on psycopg2 using sync approach and server side using sync approach and server side cursor (using support class DatabaseCursor)

Source code in src/georama/features/pygeoapi_providers/postgres.py 
 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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
class PostgresProvider(BaseProvider):
    """Generic provider for Postgresql based on psycopg2
    using sync approach and server side
    using sync approach and server side
    cursor (using support class DatabaseCursor)
    """

    def __init__(self, provider_def):
        """
        PostgreSQLProvider Class constructor

        :param provider_def: provider definitions from yml pygeoapi-config.
                             data,id_field, name set in parent class
                             data contains the connection information
                             for class DatabaseCursor

        :returns: pygeoapi.provider.base.PostgreSQLProvider
        """
        LOGGER.debug("Initialising PostgreSQL provider.")
        super().__init__(provider_def)

        self.table = provider_def["table"]
        self.id_field = provider_def["id_field"]
        self.geom = provider_def.get("geom_field", "geom")
        self.storage_crs = provider_def.get(
            "storage_crs", "https://www.opengis.net/def/crs/OGC/0/CRS84"
        )

        LOGGER.debug(f"Name: {self.name}")
        LOGGER.debug(f"Table: {self.table}")
        LOGGER.debug(f"ID field: {self.id_field}")
        LOGGER.debug(f"Geometry field: {self.geom}")

        # Read table information from database
        options = None
        if provider_def.get("options"):
            options = provider_def["options"]
        self._store_db_parameters(provider_def["data"], options)
        self._engine = get_engine(
            self.db_host,
            self.db_port,
            self.db_name,
            self.db_user,
            self._db_password,
            **(self.db_options or {}),
        )
        self.table_model = get_table_model(
            self.table, self.id_field, self.db_search_path, self._engine
        )

        LOGGER.debug(f"DB connection: {repr(self._engine.url)}")
        self.get_fields()

    def query(
        self,
        offset=0,
        limit=10,
        resulttype="results",
        bbox=[],
        datetime_=None,
        properties=[],
        sortby=[],
        select_properties=[],
        skip_geometry=False,
        q=None,
        filterq=None,
        crs_transform_spec=None,
        **kwargs,
    ):
        """
        Query Postgis for all the content.
        e,g: http://localhost:5000/collections/hotosm_bdi_waterways/items?
        limit=1&resulttype=results

        :param offset: starting record to return (default 0)
        :param limit: number of records to return (default 10)
        :param resulttype: return results or hit limit (default results)
        :param bbox: bounding box [minx,miny,maxx,maxy]
        :param datetime_: temporal (datestamp or extent)
        :param properties: list of tuples (name, value)
        :param sortby: list of dicts (property, order)
        :param select_properties: list of property names
        :param skip_geometry: bool of whether to skip geometry (default False)
        :param q: full-text search term(s)
        :param filterq: CQL query as text string
        :param crs_transform_spec: `CrsTransformSpec` instance, optional

        :returns: GeoJSON FeatureCollection
        """

        LOGGER.debug("Preparing filters")
        property_filters = self._get_property_filters(properties)
        cql_filters = self._get_cql_filters(filterq)
        bbox_filter = self._get_bbox_filter(bbox)
        time_filter = self._get_datetime_filter(datetime_)
        order_by_clauses = self._get_order_by_clauses(sortby, self.table_model)
        selected_properties = self._select_properties_clause(select_properties, skip_geometry)

        LOGGER.debug("Querying PostGIS")
        # Execute query within self-closing database Session context
        with Session(self._engine) as session:
            results = (
                session.query(self.table_model)
                .filter(property_filters)
                .filter(cql_filters)
                .filter(bbox_filter)
                .filter(time_filter)
                .options(selected_properties)
            )

            matched = results.count()

            LOGGER.debug(f"Found {matched} result(s)")

            LOGGER.debug("Preparing response")
            response = {
                "type": "FeatureCollection",
                "features": [],
                "numberMatched": matched,
                "numberReturned": 0,
            }

            if resulttype == "hits" or not results:
                return response

            crs_transform_out = self._get_crs_transform(crs_transform_spec)

            for item in (
                results.order_by(*order_by_clauses).offset(offset).limit(limit)
            ):  # noqa
                response["numberReturned"] += 1
                response["features"].append(
                    self._sqlalchemy_to_feature(item, crs_transform_out)
                )

        return response

    def get_fields(self):
        """
        Return fields (columns) from PostgreSQL table

        :returns: dict of fields
        """

        LOGGER.debug("Get available fields/properties")

        # sql-schema only allows these types, so we need to map from sqlalchemy
        # string, number, integer, object, array, boolean, null,
        # https://json-schema.org/understanding-json-schema/reference/type.html
        column_type_map = {
            bool: "boolean",
            datetime: "string",
            Decimal: "number",
            float: "number",
            int: "integer",
            str: "string",
        }
        default_type = "string"

        # https://json-schema.org/understanding-json-schema/reference/string#built-in-formats  # noqa
        column_format_map = {
            "date": "date",
            "interval": "duration",
            "time": "time",
            "timestamp": "date-time",
        }

        def _column_type_to_json_schema_type(column_type):
            try:
                python_type = column_type.python_type
            except NotImplementedError:
                LOGGER.warning(f"Unsupported column type {column_type}")
                return default_type
            else:
                try:
                    return column_type_map[python_type]
                except KeyError:
                    LOGGER.warning(f"Unsupported column type {column_type}")
                    return default_type

        def _column_format_to_json_schema_format(column_type):
            try:
                ct = str(column_type).lower()
                return column_format_map[ct]
            except KeyError:
                LOGGER.debug("No string format detected")
                return None

        if not self._fields:
            for column in self.table_model.__table__.columns:
                LOGGER.debug(f"Testing {column.name}")
                if column.name == self.geom:
                    continue

                self._fields[str(column.name)] = {
                    "type": _column_type_to_json_schema_type(column.type),
                    "format": _column_format_to_json_schema_format(column.type),
                }

        return self._fields

    def get(self, identifier, crs_transform_spec=None, **kwargs):
        """
        Query the provider for a specific
        feature id e.g: /collections/hotosm_bdi_waterways/items/13990765

        :param identifier: feature id
        :param crs_transform_spec: `CrsTransformSpec` instance, optional

        :returns: GeoJSON FeatureCollection
        """
        LOGGER.debug(f"Get item by ID: {identifier}")

        # Execute query within self-closing database Session context
        with Session(self._engine) as session:
            # Retrieve data from database as feature
            item = session.get(self.table_model, identifier)
            if item is None:
                msg = f"No such item: {self.id_field}={identifier}."
                raise ProviderItemNotFoundError(msg)
            crs_transform_out = self._get_crs_transform(crs_transform_spec)
            feature = self._sqlalchemy_to_feature(item, crs_transform_out)

            # Drop non-defined properties
            if self.properties:
                props = feature["properties"]
                dropping_keys = deepcopy(props).keys()
                for item in dropping_keys:
                    if item not in self.properties:
                        props.pop(item)

            # Add fields for previous and next items
            id_field = getattr(self.table_model, self.id_field)
            prev_item = (
                session.query(self.table_model)
                .order_by(id_field.desc())
                .filter(id_field < identifier)
                .first()
            )
            next_item = (
                session.query(self.table_model)
                .order_by(id_field.asc())
                .filter(id_field > identifier)
                .first()
            )
            feature["prev"] = (
                getattr(prev_item, self.id_field) if prev_item is not None else identifier
            )
            feature["next"] = (
                getattr(next_item, self.id_field) if next_item is not None else identifier
            )

        return feature

    def create(self, item):
        """
        Create a new item

        :param item: `dict` of new item

        :returns: identifier of created item
        """

        identifier, json_data = self._load_and_prepare_item(
            item, accept_missing_identifier=True
        )

        new_instance = self._feature_to_sqlalchemy(json_data, identifier)
        with Session(self._engine) as session:
            session.add(new_instance)
            session.commit()
            result_id = getattr(new_instance, self.id_field)

        # NOTE: need to use id from instance in case it's generated
        return result_id

    def update(self, identifier, item):
        """
        Updates an existing item

        :param identifier: feature id
        :param item: `dict` of partial or full item

        :returns: `bool` of update result
        """

        identifier, json_data = self._load_and_prepare_item(item, raise_if_exists=False)

        new_instance = self._feature_to_sqlalchemy(json_data, identifier)
        with Session(self._engine) as session:
            session.merge(new_instance)
            session.commit()

        return True

    def delete(self, identifier):
        """
        Deletes an existing item

        :param identifier: item id

        :returns: `bool` of deletion result
        """
        with Session(self._engine) as session:
            id_column = getattr(self.table_model, self.id_field)
            result = session.execute(delete(self.table_model).where(id_column == identifier))
            session.commit()

        return result.rowcount > 0

    def _store_db_parameters(self, parameters, options):
        self.db_user = parameters.get("user")
        self.db_host = parameters.get("host")
        self.db_port = parameters.get("port", 5432)
        self.db_name = parameters.get("dbname")
        # db_search_path gets converted to a tuple here in order to ensure it
        # is hashable - which allows us to use functools.cache() when
        # reflecting the table definition from the DB
        self.db_search_path = tuple(parameters.get("search_path", ["public"]))
        self._db_password = parameters.get("password")
        self.db_options = options

    def _sqlalchemy_to_feature(self, item, crs_transform_out=None):
        feature = {"type": "Feature"}

        # Add properties from item
        item_dict = item.__dict__
        item_dict.pop("_sa_instance_state")  # Internal SQLAlchemy metadata
        feature["properties"] = item_dict
        feature["id"] = item_dict.pop(self.id_field)

        # Convert geometry to GeoJSON style
        if feature["properties"].get(self.geom):
            wkb_geom = feature["properties"].pop(self.geom)
            shapely_geom = to_shape(wkb_geom)
            if crs_transform_out is not None:
                shapely_geom = crs_transform_out(shapely_geom)
            geojson_geom = shapely.geometry.mapping(shapely_geom)
            feature["geometry"] = geojson_geom
        else:
            feature["geometry"] = None

        return feature

    def _feature_to_sqlalchemy(self, json_data, identifier=None):
        attributes = {**json_data["properties"]}
        # 'identifier' key maybe be present in geojson properties, but might
        # not be a valid db field
        attributes.pop("identifier", None)
        attributes[self.geom] = from_shape(
            shapely.geometry.shape(json_data["geometry"]),
            # NOTE: for some reason, postgis in the github action requires
            # explicit crs information. i think it's valid to assume 4326:
            # https://portal.ogc.org/files/108198#feature-crs
            srid=pyproj.CRS.from_user_input(self.storage_crs).to_epsg(),
        )
        attributes[self.id_field] = identifier

        try:
            return self.table_model(**attributes)
        except Exception as e:
            LOGGER.exception("Failed to create db model")
            raise ProviderInvalidDataError(str(e))

    def _get_order_by_clauses(self, sort_by, table_model):
        # Build sort_by clauses if provided
        clauses = []
        for sort_by_dict in sort_by:
            model_column = getattr(table_model, sort_by_dict["property"])
            order_function = asc if sort_by_dict["order"] == "+" else desc
            clauses.append(order_function(model_column))

        # Otherwise sort by primary key (to ensure reproducible output)
        if not clauses:
            clauses.append(asc(getattr(table_model, self.id_field)))

        return clauses

    def _get_cql_filters(self, filterq):
        if not filterq:
            return True  # Let everything through

        # Convert filterq into SQL Alchemy filters
        field_mapping = {
            column_name: getattr(self.table_model, column_name)
            for column_name in self.table_model.__table__.columns.keys()
        }
        cql_filters = to_filter(filterq, field_mapping)

        return cql_filters

    def _get_property_filters(self, properties):
        if not properties:
            return True  # Let everything through

        # Convert property filters into SQL Alchemy filters
        # Based on https://stackoverflow.com/a/14887813/3508733
        filter_group = []
        for column_name, value in properties:
            column = getattr(self.table_model, column_name)
            filter_group.append(column == value)
        property_filters = and_(*filter_group)

        return property_filters

    def _get_bbox_filter(self, bbox):
        if not bbox:
            return True  # Let everything through

        # Convert bbx to SQL Alchemy clauses
        envelope = ST_MakeEnvelope(*bbox)
        geom_column = getattr(self.table_model, self.geom)
        bbox_filter = geom_column.intersects(envelope)

        return bbox_filter

    def _get_datetime_filter(self, datetime_):
        if datetime_ in (None, "../.."):
            return True
        else:
            if self.time_field is None:
                LOGGER.error("time_field not enabled for collection")
                raise ProviderQueryError()

            time_column = getattr(self.table_model, self.time_field)

            if "/" in datetime_:  # envelope
                LOGGER.debug("detected time range")
                time_begin, time_end = datetime_.split("/")
                if time_begin == "..":
                    datetime_filter = time_column <= time_end
                elif time_end == "..":
                    datetime_filter = time_column >= time_begin
                else:
                    datetime_filter = time_column.between(time_begin, time_end)
            else:
                datetime_filter = time_column == datetime_
        return datetime_filter

    def _select_properties_clause(self, select_properties, skip_geometry):
        # List the column names that we want
        if select_properties:
            column_names = set(select_properties)
        else:
            # get_fields() doesn't include geometry column
            column_names = set(self.fields.keys())

        if self.properties:  # optional subset of properties defined in config
            properties_from_config = set(self.properties)
            column_names = column_names.intersection(properties_from_config)

        if not skip_geometry:
            column_names.add(self.geom)

        # Convert names to SQL Alchemy clause
        selected_columns = []
        for column_name in column_names:
            try:
                column = getattr(self.table_model, column_name)
                selected_columns.append(column)
            except AttributeError:
                pass  # Ignore non-existent columns
        selected_properties_clause = load_only(*selected_columns)

        return selected_properties_clause

    def _get_crs_transform(self, crs_transform_spec=None):
        if crs_transform_spec is not None:
            crs_transform = get_transform_from_crs(
                pyproj.CRS.from_wkt(crs_transform_spec.source_crs_wkt),
                pyproj.CRS.from_wkt(crs_transform_spec.target_crs_wkt),
            )
        else:
            crs_transform = None
        return crs_transform
_engine = get_engine(self.db_host, self.db_port, self.db_name, self.db_user, self._db_password, **self.db_options or {}) instance-attribute
geom = provider_def.get('geom_field', 'geom') instance-attribute
id_field = provider_def['id_field'] instance-attribute
storage_crs = provider_def.get('storage_crs', 'https://www.opengis.net/def/crs/OGC/0/CRS84') instance-attribute
table = provider_def['table'] instance-attribute
table_model = get_table_model(self.table, self.id_field, self.db_search_path, self._engine) instance-attribute
__init__(provider_def)

PostgreSQLProvider Class constructor

:param provider_def: provider definitions from yml pygeoapi-config. data,id_field, name set in parent class data contains the connection information for class DatabaseCursor

:returns: pygeoapi.provider.base.PostgreSQLProvider

Source code in src/georama/features/pygeoapi_providers/postgres.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def __init__(self, provider_def):
    """
    PostgreSQLProvider Class constructor

    :param provider_def: provider definitions from yml pygeoapi-config.
                         data,id_field, name set in parent class
                         data contains the connection information
                         for class DatabaseCursor

    :returns: pygeoapi.provider.base.PostgreSQLProvider
    """
    LOGGER.debug("Initialising PostgreSQL provider.")
    super().__init__(provider_def)

    self.table = provider_def["table"]
    self.id_field = provider_def["id_field"]
    self.geom = provider_def.get("geom_field", "geom")
    self.storage_crs = provider_def.get(
        "storage_crs", "https://www.opengis.net/def/crs/OGC/0/CRS84"
    )

    LOGGER.debug(f"Name: {self.name}")
    LOGGER.debug(f"Table: {self.table}")
    LOGGER.debug(f"ID field: {self.id_field}")
    LOGGER.debug(f"Geometry field: {self.geom}")

    # Read table information from database
    options = None
    if provider_def.get("options"):
        options = provider_def["options"]
    self._store_db_parameters(provider_def["data"], options)
    self._engine = get_engine(
        self.db_host,
        self.db_port,
        self.db_name,
        self.db_user,
        self._db_password,
        **(self.db_options or {}),
    )
    self.table_model = get_table_model(
        self.table, self.id_field, self.db_search_path, self._engine
    )

    LOGGER.debug(f"DB connection: {repr(self._engine.url)}")
    self.get_fields()
_feature_to_sqlalchemy(json_data, identifier=None)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
def _feature_to_sqlalchemy(self, json_data, identifier=None):
    attributes = {**json_data["properties"]}
    # 'identifier' key maybe be present in geojson properties, but might
    # not be a valid db field
    attributes.pop("identifier", None)
    attributes[self.geom] = from_shape(
        shapely.geometry.shape(json_data["geometry"]),
        # NOTE: for some reason, postgis in the github action requires
        # explicit crs information. i think it's valid to assume 4326:
        # https://portal.ogc.org/files/108198#feature-crs
        srid=pyproj.CRS.from_user_input(self.storage_crs).to_epsg(),
    )
    attributes[self.id_field] = identifier

    try:
        return self.table_model(**attributes)
    except Exception as e:
        LOGGER.exception("Failed to create db model")
        raise ProviderInvalidDataError(str(e))
_get_bbox_filter(bbox)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
494
495
496
497
498
499
500
501
502
503
def _get_bbox_filter(self, bbox):
    if not bbox:
        return True  # Let everything through

    # Convert bbx to SQL Alchemy clauses
    envelope = ST_MakeEnvelope(*bbox)
    geom_column = getattr(self.table_model, self.geom)
    bbox_filter = geom_column.intersects(envelope)

    return bbox_filter
_get_cql_filters(filterq)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
467
468
469
470
471
472
473
474
475
476
477
478
def _get_cql_filters(self, filterq):
    if not filterq:
        return True  # Let everything through

    # Convert filterq into SQL Alchemy filters
    field_mapping = {
        column_name: getattr(self.table_model, column_name)
        for column_name in self.table_model.__table__.columns.keys()
    }
    cql_filters = to_filter(filterq, field_mapping)

    return cql_filters
_get_crs_transform(crs_transform_spec=None)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
555
556
557
558
559
560
561
562
563
def _get_crs_transform(self, crs_transform_spec=None):
    if crs_transform_spec is not None:
        crs_transform = get_transform_from_crs(
            pyproj.CRS.from_wkt(crs_transform_spec.source_crs_wkt),
            pyproj.CRS.from_wkt(crs_transform_spec.target_crs_wkt),
        )
    else:
        crs_transform = None
    return crs_transform
_get_datetime_filter(datetime_)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
def _get_datetime_filter(self, datetime_):
    if datetime_ in (None, "../.."):
        return True
    else:
        if self.time_field is None:
            LOGGER.error("time_field not enabled for collection")
            raise ProviderQueryError()

        time_column = getattr(self.table_model, self.time_field)

        if "/" in datetime_:  # envelope
            LOGGER.debug("detected time range")
            time_begin, time_end = datetime_.split("/")
            if time_begin == "..":
                datetime_filter = time_column <= time_end
            elif time_end == "..":
                datetime_filter = time_column >= time_begin
            else:
                datetime_filter = time_column.between(time_begin, time_end)
        else:
            datetime_filter = time_column == datetime_
    return datetime_filter
_get_order_by_clauses(sort_by, table_model)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
453
454
455
456
457
458
459
460
461
462
463
464
465
def _get_order_by_clauses(self, sort_by, table_model):
    # Build sort_by clauses if provided
    clauses = []
    for sort_by_dict in sort_by:
        model_column = getattr(table_model, sort_by_dict["property"])
        order_function = asc if sort_by_dict["order"] == "+" else desc
        clauses.append(order_function(model_column))

    # Otherwise sort by primary key (to ensure reproducible output)
    if not clauses:
        clauses.append(asc(getattr(table_model, self.id_field)))

    return clauses
_get_property_filters(properties)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
480
481
482
483
484
485
486
487
488
489
490
491
492
def _get_property_filters(self, properties):
    if not properties:
        return True  # Let everything through

    # Convert property filters into SQL Alchemy filters
    # Based on https://stackoverflow.com/a/14887813/3508733
    filter_group = []
    for column_name, value in properties:
        column = getattr(self.table_model, column_name)
        filter_group.append(column == value)
    property_filters = and_(*filter_group)

    return property_filters
_select_properties_clause(select_properties, skip_geometry)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
def _select_properties_clause(self, select_properties, skip_geometry):
    # List the column names that we want
    if select_properties:
        column_names = set(select_properties)
    else:
        # get_fields() doesn't include geometry column
        column_names = set(self.fields.keys())

    if self.properties:  # optional subset of properties defined in config
        properties_from_config = set(self.properties)
        column_names = column_names.intersection(properties_from_config)

    if not skip_geometry:
        column_names.add(self.geom)

    # Convert names to SQL Alchemy clause
    selected_columns = []
    for column_name in column_names:
        try:
            column = getattr(self.table_model, column_name)
            selected_columns.append(column)
        except AttributeError:
            pass  # Ignore non-existent columns
    selected_properties_clause = load_only(*selected_columns)

    return selected_properties_clause
_sqlalchemy_to_feature(item, crs_transform_out=None)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
def _sqlalchemy_to_feature(self, item, crs_transform_out=None):
    feature = {"type": "Feature"}

    # Add properties from item
    item_dict = item.__dict__
    item_dict.pop("_sa_instance_state")  # Internal SQLAlchemy metadata
    feature["properties"] = item_dict
    feature["id"] = item_dict.pop(self.id_field)

    # Convert geometry to GeoJSON style
    if feature["properties"].get(self.geom):
        wkb_geom = feature["properties"].pop(self.geom)
        shapely_geom = to_shape(wkb_geom)
        if crs_transform_out is not None:
            shapely_geom = crs_transform_out(shapely_geom)
        geojson_geom = shapely.geometry.mapping(shapely_geom)
        feature["geometry"] = geojson_geom
    else:
        feature["geometry"] = None

    return feature
_store_db_parameters(parameters, options)
Source code in src/georama/features/pygeoapi_providers/postgres.py 
399
400
401
402
403
404
405
406
407
408
409
def _store_db_parameters(self, parameters, options):
    self.db_user = parameters.get("user")
    self.db_host = parameters.get("host")
    self.db_port = parameters.get("port", 5432)
    self.db_name = parameters.get("dbname")
    # db_search_path gets converted to a tuple here in order to ensure it
    # is hashable - which allows us to use functools.cache() when
    # reflecting the table definition from the DB
    self.db_search_path = tuple(parameters.get("search_path", ["public"]))
    self._db_password = parameters.get("password")
    self.db_options = options
create(item)

Create a new item

:param item: dict of new item

:returns: identifier of created item

Source code in src/georama/features/pygeoapi_providers/postgres.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def create(self, item):
    """
    Create a new item

    :param item: `dict` of new item

    :returns: identifier of created item
    """

    identifier, json_data = self._load_and_prepare_item(
        item, accept_missing_identifier=True
    )

    new_instance = self._feature_to_sqlalchemy(json_data, identifier)
    with Session(self._engine) as session:
        session.add(new_instance)
        session.commit()
        result_id = getattr(new_instance, self.id_field)

    # NOTE: need to use id from instance in case it's generated
    return result_id
delete(identifier)

Deletes an existing item

:param identifier: item id

:returns: bool of deletion result

Source code in src/georama/features/pygeoapi_providers/postgres.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
def delete(self, identifier):
    """
    Deletes an existing item

    :param identifier: item id

    :returns: `bool` of deletion result
    """
    with Session(self._engine) as session:
        id_column = getattr(self.table_model, self.id_field)
        result = session.execute(delete(self.table_model).where(id_column == identifier))
        session.commit()

    return result.rowcount > 0
get(identifier, crs_transform_spec=None, **kwargs)

Query the provider for a specific feature id e.g: /collections/hotosm_bdi_waterways/items/13990765

:param identifier: feature id :param crs_transform_spec: CrsTransformSpec instance, optional

:returns: GeoJSON FeatureCollection

Source code in src/georama/features/pygeoapi_providers/postgres.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def get(self, identifier, crs_transform_spec=None, **kwargs):
    """
    Query the provider for a specific
    feature id e.g: /collections/hotosm_bdi_waterways/items/13990765

    :param identifier: feature id
    :param crs_transform_spec: `CrsTransformSpec` instance, optional

    :returns: GeoJSON FeatureCollection
    """
    LOGGER.debug(f"Get item by ID: {identifier}")

    # Execute query within self-closing database Session context
    with Session(self._engine) as session:
        # Retrieve data from database as feature
        item = session.get(self.table_model, identifier)
        if item is None:
            msg = f"No such item: {self.id_field}={identifier}."
            raise ProviderItemNotFoundError(msg)
        crs_transform_out = self._get_crs_transform(crs_transform_spec)
        feature = self._sqlalchemy_to_feature(item, crs_transform_out)

        # Drop non-defined properties
        if self.properties:
            props = feature["properties"]
            dropping_keys = deepcopy(props).keys()
            for item in dropping_keys:
                if item not in self.properties:
                    props.pop(item)

        # Add fields for previous and next items
        id_field = getattr(self.table_model, self.id_field)
        prev_item = (
            session.query(self.table_model)
            .order_by(id_field.desc())
            .filter(id_field < identifier)
            .first()
        )
        next_item = (
            session.query(self.table_model)
            .order_by(id_field.asc())
            .filter(id_field > identifier)
            .first()
        )
        feature["prev"] = (
            getattr(prev_item, self.id_field) if prev_item is not None else identifier
        )
        feature["next"] = (
            getattr(next_item, self.id_field) if next_item is not None else identifier
        )

    return feature
get_fields()

Return fields (columns) from PostgreSQL table

:returns: dict of fields

Source code in src/georama/features/pygeoapi_providers/postgres.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def get_fields(self):
    """
    Return fields (columns) from PostgreSQL table

    :returns: dict of fields
    """

    LOGGER.debug("Get available fields/properties")

    # sql-schema only allows these types, so we need to map from sqlalchemy
    # string, number, integer, object, array, boolean, null,
    # https://json-schema.org/understanding-json-schema/reference/type.html
    column_type_map = {
        bool: "boolean",
        datetime: "string",
        Decimal: "number",
        float: "number",
        int: "integer",
        str: "string",
    }
    default_type = "string"

    # https://json-schema.org/understanding-json-schema/reference/string#built-in-formats  # noqa
    column_format_map = {
        "date": "date",
        "interval": "duration",
        "time": "time",
        "timestamp": "date-time",
    }

    def _column_type_to_json_schema_type(column_type):
        try:
            python_type = column_type.python_type
        except NotImplementedError:
            LOGGER.warning(f"Unsupported column type {column_type}")
            return default_type
        else:
            try:
                return column_type_map[python_type]
            except KeyError:
                LOGGER.warning(f"Unsupported column type {column_type}")
                return default_type

    def _column_format_to_json_schema_format(column_type):
        try:
            ct = str(column_type).lower()
            return column_format_map[ct]
        except KeyError:
            LOGGER.debug("No string format detected")
            return None

    if not self._fields:
        for column in self.table_model.__table__.columns:
            LOGGER.debug(f"Testing {column.name}")
            if column.name == self.geom:
                continue

            self._fields[str(column.name)] = {
                "type": _column_type_to_json_schema_type(column.type),
                "format": _column_format_to_json_schema_format(column.type),
            }

    return self._fields
query(offset=0, limit=10, resulttype='results', bbox=[], datetime_=None, properties=[], sortby=[], select_properties=[], skip_geometry=False, q=None, filterq=None, crs_transform_spec=None, **kwargs)

Query Postgis for all the content. e,g: http://localhost:5000/collections/hotosm_bdi_waterways/items? limit=1&resulttype=results

:param offset: starting record to return (default 0) :param limit: number of records to return (default 10) :param resulttype: return results or hit limit (default results) :param bbox: bounding box [minx,miny,maxx,maxy] :param datetime_: temporal (datestamp or extent) :param properties: list of tuples (name, value) :param sortby: list of dicts (property, order) :param select_properties: list of property names :param skip_geometry: bool of whether to skip geometry (default False) :param q: full-text search term(s) :param filterq: CQL query as text string :param crs_transform_spec: CrsTransformSpec instance, optional

:returns: GeoJSON FeatureCollection

Source code in src/georama/features/pygeoapi_providers/postgres.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def query(
    self,
    offset=0,
    limit=10,
    resulttype="results",
    bbox=[],
    datetime_=None,
    properties=[],
    sortby=[],
    select_properties=[],
    skip_geometry=False,
    q=None,
    filterq=None,
    crs_transform_spec=None,
    **kwargs,
):
    """
    Query Postgis for all the content.
    e,g: http://localhost:5000/collections/hotosm_bdi_waterways/items?
    limit=1&resulttype=results

    :param offset: starting record to return (default 0)
    :param limit: number of records to return (default 10)
    :param resulttype: return results or hit limit (default results)
    :param bbox: bounding box [minx,miny,maxx,maxy]
    :param datetime_: temporal (datestamp or extent)
    :param properties: list of tuples (name, value)
    :param sortby: list of dicts (property, order)
    :param select_properties: list of property names
    :param skip_geometry: bool of whether to skip geometry (default False)
    :param q: full-text search term(s)
    :param filterq: CQL query as text string
    :param crs_transform_spec: `CrsTransformSpec` instance, optional

    :returns: GeoJSON FeatureCollection
    """

    LOGGER.debug("Preparing filters")
    property_filters = self._get_property_filters(properties)
    cql_filters = self._get_cql_filters(filterq)
    bbox_filter = self._get_bbox_filter(bbox)
    time_filter = self._get_datetime_filter(datetime_)
    order_by_clauses = self._get_order_by_clauses(sortby, self.table_model)
    selected_properties = self._select_properties_clause(select_properties, skip_geometry)

    LOGGER.debug("Querying PostGIS")
    # Execute query within self-closing database Session context
    with Session(self._engine) as session:
        results = (
            session.query(self.table_model)
            .filter(property_filters)
            .filter(cql_filters)
            .filter(bbox_filter)
            .filter(time_filter)
            .options(selected_properties)
        )

        matched = results.count()

        LOGGER.debug(f"Found {matched} result(s)")

        LOGGER.debug("Preparing response")
        response = {
            "type": "FeatureCollection",
            "features": [],
            "numberMatched": matched,
            "numberReturned": 0,
        }

        if resulttype == "hits" or not results:
            return response

        crs_transform_out = self._get_crs_transform(crs_transform_spec)

        for item in (
            results.order_by(*order_by_clauses).offset(offset).limit(limit)
        ):  # noqa
            response["numberReturned"] += 1
            response["features"].append(
                self._sqlalchemy_to_feature(item, crs_transform_out)
            )

    return response
update(identifier, item)

Updates an existing item

:param identifier: feature id :param item: dict of partial or full item

:returns: bool of update result

Source code in src/georama/features/pygeoapi_providers/postgres.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def update(self, identifier, item):
    """
    Updates an existing item

    :param identifier: feature id
    :param item: `dict` of partial or full item

    :returns: `bool` of update result
    """

    identifier, json_data = self._load_and_prepare_item(item, raise_if_exists=False)

    new_instance = self._feature_to_sqlalchemy(json_data, identifier)
    with Session(self._engine) as session:
        session.merge(new_instance)
        session.commit()

    return True

_name_for_scalar_relationship(base, local_cls, referred_cls, constraint)

Function used when automapping classes and relationships from database schema and fixes potential naming conflicts.

Source code in src/georama/features/pygeoapi_providers/postgres.py 
630
631
632
633
634
635
636
637
638
639
640
641
642
643
def _name_for_scalar_relationship(base, local_cls, referred_cls, constraint):
    """Function used when automapping classes and relationships from
    database schema and fixes potential naming conflicts.
    """
    name = referred_cls.__name__.lower()
    local_table = local_cls.__table__
    if name in local_table.columns:
        newname = name + "_"
        LOGGER.debug(
            f"Already detected column name {name!r} in table "
            f"{local_table!r}. Using {newname!r} for relationship name."
        )
        return newname
    return name

get_engine(host, port, database, user, password, **connection_options) cached

Create SQL Alchemy engine.

Source code in src/georama/features/pygeoapi_providers/postgres.py 
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
@functools.cache
def get_engine(
    host: str, port: str, database: str, user: str, password: str, **connection_options
):
    """Create SQL Alchemy engine."""
    conn_str = URL.create(
        "postgresql+psycopg2",
        username=user,
        password=password,
        host=host,
        port=int(port),
        database=database,
    )
    conn_args = {
        "client_encoding": "utf8",
        "application_name": "pygeoapi",
        **connection_options,
    }
    engine = create_engine(conn_str, connect_args=conn_args, pool_pre_ping=True)
    return engine

get_table_model(table_name, id_field, db_search_path, engine) cached

Reflect table.

Source code in src/georama/features/pygeoapi_providers/postgres.py 
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
@functools.cache
def get_table_model(
    table_name: str,
    id_field: str,
    db_search_path: tuple[str],
    engine,
):
    """Reflect table."""
    metadata = MetaData()

    # Look for table in the first schema in the search path
    schema = db_search_path[0]
    try:
        metadata.reflect(bind=engine, schema=schema, only=[table_name], views=True)
    except OperationalError:
        raise ProviderConnectionError(
            f"Could not connect to {repr(engine.url)} (password hidden)."
        )
    except InvalidRequestError:
        raise ProviderQueryError(
            f"Table '{table_name}' not found in schema '{schema}' " f"on {repr(engine.url)}."
        )

    # Create SQLAlchemy model from reflected table
    # It is necessary to add the primary key constraint because SQLAlchemy
    # requires it to reflect the table, but a view in a PostgreSQL database
    # does not have a primary key defined.
    sqlalchemy_table_def = metadata.tables[f"{schema}.{table_name}"]
    try:
        sqlalchemy_table_def.append_constraint(PrimaryKeyConstraint(id_field))
    except (ConstraintColumnNotFoundError, KeyError):
        raise ProviderQueryError(
            f"No such id_field column ({id_field}) on {schema}.{table_name}."
        )

    _Base = automap_base(metadata=metadata)
    _Base.prepare(
        name_for_scalar_relationship=_name_for_scalar_relationship,
    )
    return getattr(_Base.classes, table_name)

tests

urls

urlpatterns = [path('', views.landing_page, name='landing'), path('/conformance', views.conformance, name='conformance'), path('/openapi', views.openapi, name='openapi'), path('/collections', views.collections, name='collections'), path('/collections/<str:collection_id>', views.collections, name='collection-detail'), path('/collections/<str:collection_id>/schema', views.collection_schema, name='collection-schema'), path('/collections/<str:collection_id>/queryables', views.collection_queryables, name='collection-queryables'), path('/collections/<str:collection_id>/items', views.collection_items, name='collection-items'), path('/collections/<str:collection_id>/items/<str:item_id>', views.collection_item, name='collection-item'), path('/publish_as/oapif/<str:vector_dataset_id>', views.admin_publish_as_oapif, name='publish_as_oapif')] module-attribute

views

api = None module-attribute

admin_publish_as_oapif(request, vector_dataset_id)

helper function to hide actual connection in the database but make publishing straight forward.

Source code in src/georama/features/views.py 
416
417
418
419
420
421
422
423
def admin_publish_as_oapif(request: HttpRequest, vector_dataset_id: str):
    """
    helper function to hide actual connection in the database but make publishing straight forward.
    """
    vd = VectorDataSet.objects.filter(id=vector_dataset_id)[0]
    published_as_oapi = PublishedAsOgcApiFeatures(dataset=vd)
    published_as_oapi.save()
    return redirect("admin:features_publishedasogcapifeatures_changelist")

collection_item(request, collection_id, item_id)

OGC API collections items endpoint

:request Django HTTP Request :param collection_id: collection identifier :param item_id: item identifier

:returns: Django HTTP response

Source code in src/georama/features/views.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def collection_item(request: HttpRequest, collection_id: str, item_id: str) -> HttpResponse:
    """
    OGC API collections items endpoint

    :request Django HTTP Request
    :param collection_id: collection identifier
    :param item_id: item identifier

    :returns: Django HTTP response
    """
    published_as = PublishedAsOgcApiFeatures.objects.get(identifier=collection_id)

    if request.method == "GET":
        if published_as.has_read_permission(request.user, appname):
            response_ = execute_from_django(
                itemtypes_api.get_collection_item, request, collection_id, item_id
            )
        else:
            raise PermissionDenied()
    elif request.method == "PUT":
        if published_as.has_update_permission(request.user, appname):
            response_ = execute_from_django(
                itemtypes_api.manage_collection_item,
                request,
                "update",
                collection_id,
                item_id,
                skip_valid_check=True,
            )
        else:
            raise PermissionDenied()
    elif request.method == "DELETE":
        if published_as.has_delete_permission(request.user, appname):
            response_ = execute_from_django(
                itemtypes_api.manage_collection_item,
                request,
                "delete",
                collection_id,
                item_id,
                skip_valid_check=True,
            )
        else:
            raise PermissionDenied()
    elif request.method == "OPTIONS":
        if published_as.has_read_permission(request.user, appname):
            response_ = execute_from_django(
                itemtypes_api.manage_collection_item,
                request,
                "options",
                collection_id,
                item_id,
                skip_valid_check=True,
            )
        else:
            raise PermissionDenied()
    else:
        raise BadRequest()

    return response_

collection_items(request, collection_id)

OGC API collections items endpoint

:request Django HTTP Request :param collection_id: collection identifier

:returns: Django HTTP response

Source code in src/georama/features/views.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def collection_items(request: HttpRequest, collection_id: str) -> HttpResponse:
    """
    OGC API collections items endpoint

    :request Django HTTP Request
    :param collection_id: collection identifier

    :returns: Django HTTP response
    """

    published_as = PublishedAsOgcApiFeatures.objects.get(identifier=collection_id)

    if request.method == "GET":
        if published_as.has_read_permission(request.user, appname):
            response_ = execute_from_django(
                itemtypes_api.get_collection_items,
                request,
                collection_id,
                skip_valid_check=True,
            )
        else:
            raise PermissionDenied()
    elif request.method == "POST":
        if published_as.has_create_permission(request.user, appname):
            if request.content_type is not None:
                if request.content_type == "application/geo+json":
                    response_ = execute_from_django(
                        itemtypes_api.manage_collection_item,
                        request,
                        "create",
                        collection_id,
                        skip_valid_check=True,
                    )
                else:
                    response_ = execute_from_django(
                        itemtypes_api.post_collection_items,
                        request,
                        collection_id,
                        skip_valid_check=True,
                    )
        else:
            raise PermissionDenied()
    elif request.method == "OPTIONS":
        if published_as.has_read_permission(request.user, appname):
            response_ = execute_from_django(
                itemtypes_api.manage_collection_item,
                request,
                "options",
                collection_id,
                skip_valid_check=True,
            )
        else:
            raise PermissionDenied()
    else:
        raise BadRequest()

    return response_

collection_queryables(request, collection_id=None)

OGC API collections queryables endpoint

:request Django HTTP Request :param collection_id: collection identifier

:returns: Django HTTP Response

Source code in src/georama/features/views.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def collection_queryables(
    request: HttpRequest, collection_id: typing.Optional[str] = None
) -> HttpResponse:
    """
    OGC API collections queryables endpoint

    :request Django HTTP Request
    :param collection_id: collection identifier

    :returns: Django HTTP Response
    """
    if PublishedAsOgcApiFeatures.objects.get(identifier=collection_id).has_read_permission(
        request.user, appname
    ):
        return execute_from_django(
            itemtypes_api.get_collection_queryables, request, collection_id
        )
    else:
        raise PermissionDenied()

collection_schema(request, collection_id=None)

OGC API collections schema endpoint

:request Django HTTP Request :param collection_id: collection identifier

:returns: Django HTTP Response

Source code in src/georama/features/views.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def collection_schema(
    request: HttpRequest, collection_id: typing.Optional[str] = None
) -> HttpResponse:
    """
    OGC API collections schema endpoint

    :request Django HTTP Request
    :param collection_id: collection identifier

    :returns: Django HTTP Response
    """
    if PublishedAsOgcApiFeatures.objects.get(identifier=collection_id).has_read_permission(
        request.user, appname
    ):
        return execute_from_django(core_api.get_collection_schema, request, collection_id)
    else:
        raise PermissionDenied()

collections(request, collection_id=None)

OGC API collections endpoint

:request Django HTTP Request :param collection_id: collection identifier

:returns: Django HTTP Response

Source code in src/georama/features/views.py 
56
57
58
59
60
61
62
63
64
65
66
67
def collections(
    request: HttpRequest, collection_id: typing.Optional[str] = None
) -> HttpResponse:
    """
    OGC API collections endpoint

    :request Django HTTP Request
    :param collection_id: collection identifier

    :returns: Django HTTP Response
    """
    return execute_from_django(core_api.describe_collections, request, collection_id)

conformance(request)

OGC API conformance endpoint

:request Django HTTP Request

:returns: Django HTTP Response

Source code in src/georama/features/views.py 
45
46
47
48
49
50
51
52
53
def conformance(request: HttpRequest) -> HttpResponse:
    """
    OGC API conformance endpoint

    :request Django HTTP Request

    :returns: Django HTTP Response
    """
    return execute_from_django(core_api.conformance, request)

create_ogr_provider(published_as, editable, features_properties)

Source code in src/georama/features/views.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def create_ogr_provider(
    published_as: PublishedAsOgcApiFeatures,
    editable: bool,
    features_properties: typing.List[str],
) -> dict:
    source, path = published_as.dataset.source_to_qsl
    crs = published_as.dataset.crs_to_qsl
    handle_crs_setting(crs.ogc_uri),
    config = Config()
    # TODO: make this configurable
    driver_lookup = {"SHP": "ESRI Shapefile", "GPKG": "GPKG", "GDB": "OpenFileGDB"}
    available_crs_list = [crs.ogc_uri, "https://www.opengis.net/def/crs/OGC/0/CRS84"]
    if config.default_crs not in available_crs_list:
        available_crs_list.append(config.default_crs)
    provider_definition = {
        "type": "feature",
        "name": "OGR",
        "data": {
            "source_type": driver_lookup[source.ogr.path.split(".")[-1].upper()],
            "source": os.path.join(
                config.path, published_as.dataset.project.mandant.name, source.ogr.path
            ),
            "source_capabilities": {"paging": True},
        },
        "editable": editable,
        "crs": available_crs_list,
        "storage_crs": crs.ogc_uri,
        "id_field": "fid",
        "layer": source.ogr.layer_name
        if source.ogr.layer_name is not None
        else os.path.basename(source.ogr.path).split(".")[0],
        # TODO:
        # "id_field": "fid",
        # "title_field": "kantonsname",
    }
    if published_as.column_permission:
        if len(features_properties) == 0:
            # for OGR `geom` is the standard geometry column
            features_properties = ["geom"]
        provider_definition["properties"] = features_properties

    return provider_definition

create_postgres_provider(published_as, editable, features_properties)

Source code in src/georama/features/views.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def create_postgres_provider(
    published_as: PublishedAsOgcApiFeatures,
    editable: bool,
    features_properties: typing.List[str],
) -> dict:
    source, path = published_as.dataset.source_to_qsl
    crs = published_as.dataset.crs_to_qsl
    handle_crs_setting(crs.ogc_uri)

    available_crs_list = [crs.ogc_uri, "https://www.opengis.net/def/crs/OGC/0/CRS84"]

    if Config().default_crs not in available_crs_list:
        available_crs_list.append(Config().default_crs)

    provider_definition = {
        "type": "feature",
        "name": "OG_POSTGRES",
        "data": {
            "host": source.postgres.host,
            "port": source.postgres.port,
            "dbname": source.postgres.dbname,
            "user": source.postgres.username,
            "password": source.postgres.password,
            "search_path": [source.postgres.schema],
        },
        "editable": editable,
        "crs": available_crs_list,
        "storage_crs": crs.ogc_uri,
        "id_field": source.postgres.key,
        "table": source.postgres.table,
        "geom_field": source.postgres.geometry_column
        # TODO:
        # "id_field": "fid",
        # "title_field": "kantonsname",
    }
    if published_as.column_permission:
        if len(features_properties) == 0:
            features_properties = [source.postgres.geometry_column]
        provider_definition["properties"] = features_properties

    return provider_definition

create_resource(published_as, request)

Source code in src/georama/features/views.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
def create_resource(published_as: PublishedAsOgcApiFeatures, request: HttpRequest) -> dict:
    editable = (
        published_as.has_update_permission(request.user, appname)
        or published_as.has_create_permission(request.user, appname)
        or published_as.has_delete_permission(request.user, appname)
    )

    features_properties = []
    if published_as.column_permission:
        features_properties = [
            c.name
            for c in published_as.columns.all()
            if c.has_general_permission(request.user, appname)
        ]

    if published_as.dataset.driver.upper() == "POSTGRES":
        provider = create_postgres_provider(published_as, editable, features_properties)
    elif published_as.dataset.driver.upper() == "OGR":
        provider = create_ogr_provider(published_as, editable, features_properties)
    else:
        raise NotImplementedError

    return {
        "type": "collection",
        "title": published_as.title,
        "description": published_as.description,
        # TODO: add keywords into models
        "keywords": [],
        "linked-data": {
            "context": [
                {"datetime": "https://schema.org/DateTime"},
                {
                    "vocab": "https://example.com/vocab#",
                    "stn_id": "vocab:stn_id",
                    "value": "vocab:value",
                },
            ]
        },
        "links": [],
        "extents": {
            "spatial": {
                "bbox": BBox.from_string(published_as.dataset.bbox).to_list(),
                "crs": "http://www.opengis.net/def/crs/OGC/1.3/CRS84",
            }
        },
        "providers": [provider],
        "limits": {
            "on_exceed": published_as.on_exceed,
            "max_items": published_as.max_items,
            "default_items": published_as.default_items,
        },
    }

execute_from_django(api_function, request, *args, skip_valid_check=False)

Source code in src/georama/features/views.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def execute_from_django(
    api_function, request: HttpRequest, *args, skip_valid_check=False
) -> HttpResponse:
    # TODO: This has to be stored somewhere, maybe a session store might be good
    server_config, openapi_config = handle_runtime_config(request)
    api = API(server_config, openapi_config)
    # TODO: this only needs to be done once the config actually changes aka published features are added or
    #       deleted!
    l10n._cfg_cache = {}

    api_request = APIRequest.from_django(request, api.locales)
    content: typing.Union[str, bytes]
    if not skip_valid_check and not api_request.is_valid():
        headers, status, content = api.get_format_exception(api_request)
    else:

        headers, status, content = api_function(api, api_request, *args)
        content = apply_gzip(headers, content)

    # Convert API payload to a django response
    response = HttpResponse(content, status=status)

    for key, value in headers.items():
        response[key] = value
    return response

handle_crs_setting(crs)

Source code in src/georama/features/views.py 
268
269
270
271
272
def handle_crs_setting(crs: str):
    import pygeoapi.api as runtime_api

    if crs not in runtime_api.DEFAULT_CRS_LIST:
        runtime_api.DEFAULT_CRS_LIST.append(crs)

handle_runtime_config(request)

Source code in src/georama/features/views.py 
230
231
232
233
234
235
236
237
238
def handle_runtime_config(request: HttpRequest) -> tuple[dict, dict]:
    server_config = ServerConfig().get()
    server_config["server"]["url"] = f"{request.scheme}://{request.get_host()}/features"
    for published_as in PublishedAsOgcApiFeatures.objects.all():
        if published_as.has_general_permission(request.user, appname):
            server_config["resources"][str(published_as.identifier)] = create_resource(
                published_as, request
            )
    return server_config, get_oas(server_config)

landing_page(request)

OGC API landing page endpoint

:request Django HTTP Request

:returns: Django HTTP Response

Source code in src/georama/features/views.py 
23
24
25
26
27
28
29
30
31
def landing_page(request: HttpRequest) -> HttpResponse:
    """
    OGC API landing page endpoint

    :request Django HTTP Request

    :returns: Django HTTP Response
    """
    return execute_from_django(core_api.landing_page, request)

openapi(request)

OpenAPI endpoint

:request Django HTTP Request

:returns: Django HTTP Response

Source code in src/georama/features/views.py 
34
35
36
37
38
39
40
41
42
def openapi(request: HttpRequest) -> HttpResponse:
    """
    OpenAPI endpoint

    :request Django HTTP Request

    :returns: Django HTTP Response
    """
    return execute_from_django(core_api.openapi_, request)