huanld/pygeoapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixes and adjustments for the Oracle provider (#1410)

Browse Source 
* Added support for table synonyms

* Added new parameters to query and manipulator call

* Changed error types

* Mount volumes to oracle container

* workflow part 2

* workflow part 3

* workflow part 4

* Changed file permissions to 777

* Deleted folder

* Recreated folder

* Changed to official Oracle Docker-Image

* Added Chown user

* back to gvenzl/oracle-xe:latest

* Tried docker-entrypoint-startdb.d

* Added addnab/docker-run-action@v3

* Added port and deamon mode

* next try

* added job.container.network

* next try

* + docker ps

* next try

* using docker run

* next try

* next try

* Changed len of array to 11

* Use sdo_util.from_geojsonfor create and update

* Flake8 changes

* Fixed error with views

* Added crs_transform_spec support

* Without default_crs

* Updated documentation for Oracle Provider

* changes for flake8

* Added crs_transform_spec support to get function

* review changes

* Added configurable SDO operator
This commit is contained in:
xkosubek
2023-12-16 13:21:25 +01:00
committed by GitHub
parent d25d0cfa65
commit f5b6505b67
7 changed files with 506 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/main.yml
+7 -20
View File
@@ -43,27 +43,10 @@ jobs:
env:
PYGEOAPI_CONFIG: "$(pwd)/pygeoapi-config.yml"
services:
# Oracle service (label used to access the service container)
oracle:
# Docker Hub image (feel free to change the tag "latest" to any other available one)
image: gvenzl/oracle-xe:latest
# Provide passwords and other environment variables to container
env:
ORACLE_RANDOM_PASSWORD: true
APP_USER: geo_test
APP_USER_PASSWORD: geo_test
# Forward Oracle port
ports:
- 1521:1521
# Provide healthcheck script options for startup
options: >-
--health-cmd healthcheck.sh
--health-interval 10s
--health-timeout 5s
--health-retries 10
steps:
- name: Chown user
run: |
sudo chown -R $USER:$USER $GITHUB_WORKSPACE
- uses: actions/checkout@v2
- uses: actions/setup-python@v2
name: Setup Python ${{ matrix.python-version }}
@@ -105,6 +88,9 @@ jobs:
with:
packages: gdal-bin libgdal-dev
version: 3.0.4
- name: Install and run Oracle
run: |
docker run -d --name oracledb -e ORACLE_PWD=oracle -v ${{ github.workspace }}/tests/data/oracle/init-db:/opt/oracle/scripts/startup -p 1521:1521 container-registry.oracle.com/database/express:21.3.0-xe
- name: Install requirements 📦
run: |
pip3 install -r requirements.txt
@@ -125,6 +111,7 @@ jobs:
python3 tests/load_mongo_data.py tests/data/ne_110m_populated_places_simple.geojson
gunzip < tests/data/hotosm_bdi_waterways.sql.gz | psql postgresql://postgres:${{ secrets.DatabasePassword || 'postgres' }}@localhost:5432/test
psql postgresql://postgres:${{ secrets.DatabasePassword || 'postgres' }}@localhost:5432/test -f tests/data/dummy_data.sql
docker ps
python3 tests/load_oracle_data.py
- name: run unit tests ⚙️
env:
docs/source/data-publishing/ogcapi-features.rst
+54 -8
View File
@@ -26,6 +26,7 @@ parameters.
`GeoJSON`_,✅/✅,results/hits,❌,❌,❌,✅,❌,❌,✅
`MongoDB`_,✅/❌,results,✅,✅,✅,✅,❌,❌,✅
`OGR`_,✅/❌,results/hits,✅,❌,❌,✅,❌,❌,✅
`Oracle`_,✅/✅,results/hits,✅,❌,✅,✅,❌,❌,✅
`PostgreSQL`_,✅/✅,results/hits,✅,✅,✅,✅,✅,❌,✅
`SQLiteGPKG`_,✅/❌,results/hits,✅,❌,❌,✅,❌,❌,✅
`SensorThings API`_,✅/✅,results/hits,✅,✅,✅,✅,❌,❌,✅
@@ -274,6 +275,8 @@ Oracle
.. note::
Requires Python package oracledb
Connection
""""""""""
.. code-block:: yaml
providers:
@@ -295,21 +298,64 @@ Oracle
table: lakes
geom_field: geometry
title_field: name
# sql_manipulator: tests.test_oracle_provider.SqlManipulator
# sql_manipulator_options:
# foo: bar
# mandatory_properties:
# - bbox
# source_crs: 31287 # defaults to 4326 if not provided
# target_crs: 31287 # defaults to 4326 if not provided
The provider supports connection over host and port with SID or SERVICE_NAME. For TNS naming, the system
The provider supports connection over host and port with SID, SERVICE_NAME or TNS_NAME. For TNS naming, the system
environment variable TNS_ADMIN or the configuration parameter tns_admin must be set.
The providers supports external authentication. At the moment only wallet authentication is implemented.
Sometimes it is necessary to use the Oracle client for the connection. In this case init_oracle_client must be set to True.
SDO options
"""""""""""
.. code-block:: yaml
providers:
- type: feature
name: OracleDB
data:
host: 127.0.0.1
port: 1521
service_name: XEPDB1
user: geo_test
password: geo_test
id_field: id
table: lakes
geom_field: geometry
title_field: name
sdo_operator: sdo_relate # defaults to sdo_filter
sdo_param: mask=touch+coveredby # defaults to mask=anyinteract
The provider supports two different SDO operators, sdo_filter and sdo_relate. When not set, the default is sdo_relate!
Further more it is possible to set the sdo_param option. When sdo_relate is used the default is anyinteraction!
`See Oracle Documentation for details <https://docs.oracle.com/en/database/oracle/oracle-database/23/spatl/spatial-operators-reference.html>`_.
Mandatory properties
""""""""""""""""""""
.. code-block:: yaml
providers:
- type: feature
name: OracleDB
data:
host: 127.0.0.1
port: 1521
service_name: XEPDB1
user: geo_test
password: geo_test
id_field: id
table: lakes
geom_field: geometry
title_field: name
manadory_properties:
- example_group_id
On large tables it could be useful to disallow a query on the complete dataset. For this reason it is possible to
configure mandatory properties. When this is activated, the provoder throws an exception when the parameter
is not in the query uri.
Custom SQL Manipulator Plugin
"""""""""""""""""""""""""""""
The provider supports a SQL-Manipulator-Plugin class. With this, the SQL statement could be manipulated. This is
useful e.g. for authorization at row level or manipulation of the explain plan with hints.
pygeoapi/provider/oracle.py
+321 -95
View File
@@ -30,17 +30,23 @@
import importlib
import json
import logging
import oracledb
import pyproj
from typing import Optional
import oracledb
from pygeoapi.api import DEFAULT_STORAGE_CRS
from pygeoapi.provider.base import (
BaseProvider,
ProviderConnectionError,
ProviderGenericError,
ProviderInvalidQueryError,
ProviderItemNotFoundError,
ProviderQueryError,
)
from pygeoapi.util import get_crs_from_uri
LOGGER = logging.getLogger(__name__)
@@ -97,7 +103,7 @@ class DatabaseConnection:
)
if "tns_name" not in self.conn_dict:
raise Exception(
raise ProviderConnectionError(
"tns_name must be set for external authentication!"
)
@@ -111,7 +117,7 @@ class DatabaseConnection:
)
if "host" not in self.conn_dict:
raise Exception(
raise ProviderConnectionError(
"Host must be set for connection with service_name!"
)
@@ -128,7 +134,7 @@ class DatabaseConnection:
)
if "host" not in self.conn_dict:
raise Exception(
raise ProviderConnectionError(
"Host must be set for connection with sid!"
)
@@ -184,7 +190,8 @@ class DatabaseConnection:
LOGGER.error(e)
raise ProviderConnectionError(e)
# Check if table name has schema inside
# Check if table name has schema/owner inside
# If not, current user is set
table_parts = self.table.split(".")
if len(table_parts) == 2:
schema = table_parts[0]
@@ -196,34 +203,23 @@ class DatabaseConnection:
LOGGER.debug("Schema: " + schema)
LOGGER.debug("Table: " + table)
self.cur = self.conn.cursor()
if self.context == "query":
# Get table column names and types, excluding geometry
query_cols = "select column_name, data_type \
from all_tab_columns \
where table_name = UPPER(:table_name) \
and owner = UPPER(:owner) \
and data_type != 'SDO_GEOMETRY'"
self.cur.execute(
query_cols, {"table_name": table, "owner": schema}
)
result = self.cur.fetchall()
column_list = self._get_table_columns(schema, table)
# When self.properties is set, then the result would be filtered
if self.properties:
result = [
res
for res in result
if res[0].lower()
column_list = [
col
for col in column_list
if col[0].lower()
in [item.lower() for item in self.properties]
]
# Concatenate column names with ', '
self.columns = ", ".join([item[0].lower() for item in result])
self.columns = ", ".join([item[0].lower() for item in column_list])
# Populate dictionary for columns with column type
for k, v in dict(result).items():
for k, v in dict(column_list).items():
self.fields[k.lower()] = {"type": v}
return self
@@ -232,6 +228,82 @@ class DatabaseConnection:
# some logic to commit/rollback
self.conn.close()
def _get_table_columns(self, schema, table):
"""
Returns an array with all column names and data types
from Oracle table ALL_TAB_COLUMNS.
Lookup for public and private synonyms.
Throws ProviderGenericError when table not exist or accesable.
"""
sql = """
SELECT COUNT(1)
FROM all_objects
WHERE object_type IN ('VIEW','TABLE','MATERIALIZED VIEW')
AND object_name = UPPER(:table_name)
AND owner = UPPER(:owner)
"""
with self.conn.cursor() as cur:
cur.execute(sql, {"table_name": table, "owner": schema})
result = cur.fetchone()
if result[0] == 0:
sql = """
SELECT COUNT(1)
FROM all_synonyms
WHERE synonym_name = UPPER(:table_name)
AND owner = UPPER(:owner)
"""
with self.conn.cursor() as cur:
cur.execute(sql, {"table_name": table, "owner": schema})
result = cur.fetchone()
if result[0] == 0:
sql = """
SELECT COUNT(1)
FROM all_synonyms
WHERE synonym_name = UPPER(:table_name)
AND owner = 'PUBLIC'
"""
with self.conn.cursor() as cur:
cur.execute(sql, {"table_name": table})
result = cur.fetchone()
if result[0] == 0:
raise ProviderGenericError(
f"Table {schema}.{table} not found!"
)
else:
schema = "PUBLIC"
sql = """
SELECT table_owner, table_name
FROM all_synonyms
WHERE synonym_name = UPPER(:table_name)
AND owner = UPPER(:owner)
"""
with self.conn.cursor() as cur:
cur.execute(sql, {"table_name": table, "owner": schema})
result = cur.fetchone()
schema = result[0]
table = result[1]
# Get table column names and types, excluding geometry
query_cols = """
SELECT column_name, data_type
FROM all_tab_columns
WHERE table_name = UPPER(:table_name)
AND owner = UPPER(:owner)
AND data_type != 'SDO_GEOMETRY'
"""
with self.conn.cursor() as cur:
cur.execute(query_cols, {"table_name": table, "owner": schema})
result = cur.fetchall()
return result
class OracleProvider(BaseProvider):
def __init__(self, provider_def):
@@ -248,28 +320,42 @@ class OracleProvider(BaseProvider):
super().__init__(provider_def)
# Table properties
self.table = provider_def["table"]
self.id_field = provider_def["id_field"]
self.conn_dic = provider_def["data"]
self.geom = provider_def["geom_field"]
self.properties = [item.lower() for item in self.properties]
self.mandatory_properties = provider_def.get("mandatory_properties")
# SQL manipulator properties
self.sql_manipulator = provider_def.get("sql_manipulator")
self.sql_manipulator_options = provider_def.get(
"sql_manipulator_options"
)
self.mandatory_properties = provider_def.get("mandatory_properties")
self.source_crs = provider_def.get("source_crs", 4326)
self.target_crs = provider_def.get("target_crs", 4326)
self.sdo_mask = provider_def.get("sdo_mask", "anyinteraction")
# CRS properties
storage_crs_uri = provider_def.get("storage_crs", DEFAULT_STORAGE_CRS)
self.storage_crs = get_crs_from_uri(storage_crs_uri)
# TODO See Issue #1393
# default_crs_uri = provider_def.get("default_crs", DEFAULT_CRS)
# self.default_crs = get_crs_from_uri(default_crs_uri)
# SDO properties
self.sdo_param = provider_def.get("sdo_param")
self.sdo_operator = provider_def.get("sdo_operator", "sdo_filter")
LOGGER.debug("Setting Oracle properties:")
LOGGER.debug(f"Name:{self.name}")
LOGGER.debug(f"ID_field:{self.id_field}")
LOGGER.debug(f"Table:{self.table}")
LOGGER.debug(f"source_crs: {self.source_crs}")
LOGGER.debug(f"target_crs: {self.target_crs}")
LOGGER.debug(f"sdo_mask: {self.sdo_mask}")
LOGGER.debug(f"sdo_param: {self.sdo_param}")
LOGGER.debug(f"sdo_operator: {self.sdo_operator}")
LOGGER.debug(f"storage_crs {self.storage_crs}")
# TODO See Issue #1393
# LOGGER.debug(f"default_crs: {self.default_crs}")
self.get_fields()
@@ -289,7 +375,12 @@ class OracleProvider(BaseProvider):
return self.fields
def _get_where_clauses(
self, properties, bbox, bbox_crs, sdo_mask="anyinteraction"
self,
properties,
bbox,
bbox_crs,
sdo_param=None,
sdo_operator="sdo_filter",
):
"""
Generarates WHERE conditions to be implemented in query.
@@ -313,33 +404,72 @@ class OracleProvider(BaseProvider):
if bbox:
bbox_dict = {"clause": "", "properties": {}}
sdo_mask = f"mask={sdo_mask}"
if sdo_operator == "sdo_relate":
if not sdo_param:
sdo_param = "mask=anyinteract"
bbox_dict["properties"] = {
"srid": bbox_crs or 4326,
"minx": bbox[0],
"miny": bbox[1],
"maxx": bbox[2],
"maxy": bbox[3],
"sdo_mask": sdo_mask,
}
bbox_dict["properties"] = {
"srid": self._get_srid_from_crs(bbox_crs),
"minx": bbox[0],
"miny": bbox[1],
"maxx": bbox[2],
"maxy": bbox[3],
"sdo_param": sdo_param,
}
bbox_dict[
"clause"
] = f"sdo_relate({self.geom}, \
mdsys.sdo_geometry(2003, \
:srid, \
NULL, \
mdsys.sdo_elem_info_array(\
1, \
1003, \
3\
), \
mdsys.sdo_ordinate_array(:minx, \
:miny, \
:maxx, \
:maxy)), \
:sdo_mask) = 'TRUE'"
bbox_query = f"""
sdo_relate({self.geom},
mdsys.sdo_geometry(2003,
:srid,
NULL,
mdsys.sdo_elem_info_array(
1,
1003,
3
),
mdsys.sdo_ordinate_array(
:minx,
:miny,
:maxx,
:maxy
)
),
:sdo_param
) = 'TRUE'
"""
else:
bbox_dict["properties"] = {
"srid": self._get_srid_from_crs(bbox_crs),
"minx": bbox[0],
"miny": bbox[1],
"maxx": bbox[2],
"maxy": bbox[3],
"sdo_param": sdo_param,
}
bbox_query = f"""
sdo_filter({self.geom},
mdsys.sdo_geometry(2003,
:srid,
NULL,
mdsys.sdo_elem_info_array(
1,
1003,
3
),
mdsys.sdo_ordinate_array(
:minx,
:miny,
:maxx,
:maxy
)
),
:sdo_param
) = 'TRUE'
"""
bbox_dict["clause"] = bbox_query
where_conditions.append(bbox_dict["clause"])
where_dict["properties"].update(bbox_dict["properties"])
@@ -381,6 +511,20 @@ class OracleProvider(BaseProvider):
oracledb.DB_TYPE_LONG_RAW, arraysize=cursor.arraysize
)
def _get_srid_from_crs(self, crs):
"""
Works only for EPSG codes!
Anything else is hard coded!
"""
if crs == "OGC:CRS84":
srid = 4326
elif crs == "OGC:CRS84h":
srid = 4326
else:
srid = crs.to_epsg()
return srid
def query(
self,
offset=0,
@@ -390,9 +534,11 @@ class OracleProvider(BaseProvider):
datetime_=None,
properties=[],
sortby=[],
select_properties=[],
skip_geometry=False,
select_properties=[],
crs_transform_spec=None,
q=None,
language=None,
filterq=None,
**kwargs,
):
@@ -419,12 +565,12 @@ class OracleProvider(BaseProvider):
if self.mandatory_properties:
for mand_col in self.mandatory_properties:
if mand_col == "bbox" and not bbox:
raise ProviderQueryError(
raise ProviderInvalidQueryError(
f"Missing mandatory filter property: {mand_col}"
)
else:
if mand_col not in property_dict:
raise ProviderQueryError(
raise ProviderInvalidQueryError(
f"Missing mandatory filter property: {mand_col}"
)
@@ -440,8 +586,9 @@ class OracleProvider(BaseProvider):
where_dict = self._get_where_clauses(
properties=properties,
bbox=bbox,
bbox_crs=self.source_crs,
sdo_mask=self.sdo_mask,
bbox_crs=self.storage_crs,
sdo_param=self.sdo_param,
sdo_operator=self.sdo_operator,
)
# Not dangerous to use self.table as substitution,
@@ -481,26 +628,49 @@ class OracleProvider(BaseProvider):
where_dict = self._get_where_clauses(
properties=properties,
bbox=bbox,
bbox_crs=self.source_crs,
sdo_mask=self.sdo_mask,
bbox_crs=self.storage_crs,
sdo_param=self.sdo_param,
sdo_operator=self.sdo_operator,
)
# Get correct SRID
if crs_transform_spec is not None:
source_crs = pyproj.CRS.from_wkt(
crs_transform_spec.source_crs_wkt
)
source_srid = self._get_srid_from_crs(source_crs)
target_crs = pyproj.CRS.from_wkt(
crs_transform_spec.target_crs_wkt
)
target_srid = self._get_srid_from_crs(target_crs)
else:
source_srid = self._get_srid_from_crs(self.storage_crs)
target_srid = source_srid
# TODO See Issue #1393
# target_srid = self._get_srid_from_crs(self.default_crs)
# If issue is not accepted, this block can be merged with
# the following block.
LOGGER.debug(f"source_srid: {source_srid}")
LOGGER.debug(f"target_srid: {target_srid}")
# Build geometry column call
# When a different output CRS is definded, the geometry
# geometry column would be transformed.
if skip_geometry:
geom = ""
elif (
not skip_geometry
and self.target_crs
and self.target_crs != self.source_crs
):
geom = f", sdo_cs.transform(t1.{self.geom}, \
:target_srid).get_geojson() \
AS geometry "
elif source_srid != target_srid:
geom = f""", sdo_cs.transform(t1.{self.geom},
:target_srid).get_geojson()
AS geometry """
where_dict["properties"].update(
{"target_srid": int(self.target_crs)}
{"target_srid": int(target_srid)}
)
else:
geom = f", t1.{self.geom}.get_geojson() AS geometry "
@@ -534,9 +704,19 @@ class OracleProvider(BaseProvider):
sql_query,
bind_variables,
self.sql_manipulator_options,
offset,
limit,
resulttype,
bbox,
self.source_crs,
datetime_,
properties,
sortby,
skip_geometry,
select_properties,
crs_transform_spec,
q,
language,
filterq,
)
# Clean up placeholders that aren't used by the
@@ -622,7 +802,7 @@ class OracleProvider(BaseProvider):
return id
def get(self, identifier, **kwargs):
def get(self, identifier, crs_transform_spec=None, **kwargs):
"""
Query the provider for a specific
feature id e.g: /collections/ocrl_lakes/items/1
@@ -640,11 +820,41 @@ class OracleProvider(BaseProvider):
cursor = db.conn.cursor()
crs_dict = {}
if self.target_crs and self.target_crs != self.source_crs:
geom_sql = f", sdo_cs.transform(t1.{self.geom}, \
:target_srid).get_geojson() \
AS geometry "
crs_dict = {"target_srid": int(self.target_crs)}
# Get correct SRIDs
if crs_transform_spec is not None:
source_crs = pyproj.CRS.from_wkt(
crs_transform_spec.source_crs_wkt
)
source_srid = self._get_srid_from_crs(source_crs)
target_crs = pyproj.CRS.from_wkt(
crs_transform_spec.target_crs_wkt
)
target_srid = self._get_srid_from_crs(target_crs)
else:
source_srid = self._get_srid_from_crs(self.storage_crs)
target_srid = source_srid
# TODO See Issue #1393
# target_srid = self._get_srid_from_crs(self.default_crs)
# If issue is not accepted, this block can be merged with
# the following block.
LOGGER.debug(f"source_srid: {source_srid}")
LOGGER.debug(f"target_srid: {target_srid}")
# Build geometry column call
# When a different output CRS is definded, the geometry
# geometry column would be transformed.
if source_srid != target_srid:
crs_dict = {"target_srid": target_srid}
geom_sql = f""", sdo_cs.transform(t1.{self.geom},
:target_srid).get_geojson()
AS geometry """
else:
geom_sql = f", t1.{self.geom}.get_geojson() AS geometry "
@@ -779,24 +989,32 @@ class OracleProvider(BaseProvider):
columns_str = ", ".join([col for col in columns])
values_str = ", ".join([f":{col}" for col in columns])
sql_query = f"INSERT INTO {self.table} (\
{columns_str}, \
{self.geom}) \
VALUES ({values_str}, :in_geometry) \
RETURNING {self.id_field} INTO :out_id"
sql_query = f"""
INSERT INTO {self.table} (
{columns_str},
{self.geom}
)
VALUES (
{values_str},
sdo_util.from_geojson(:in_geometry, NULL, :srid)
)
RETURNING {self.id_field} INTO :out_id
"""
# Out bind variable for the id of the created row
out_id = cursor.var(int)
# Bind variable for the SDO_GEOMETRY type
in_geometry = self._get_sdo_from_geojson_geometry(
db.conn, request_data.get("geometry").get("coordinates")[0]
)
# in_geometry = self._get_sdo_from_geojson_geometry(
# db.conn, request_data.get("geometry").get("coordinates")[0]
# )
in_geometry = request_data.get("geometry")
bind_variables = {
**bind_variables,
"out_id": out_id,
"in_geometry": in_geometry,
"in_geometry": json.dumps(in_geometry),
"srid": self._get_srid_from_crs(self.storage_crs),
}
# SQL manipulation plugin
@@ -872,20 +1090,28 @@ class OracleProvider(BaseProvider):
set_str = ", ".join([f" {col} = :{col}" for col in columns])
sql_query = f"UPDATE {self.table} \
SET {set_str} \
, {self.geom} = :in_geometry \
WHERE {self.id_field} = :in_id"
sql_query = f"""
UPDATE {self.table}
SET {set_str}
, {self.geom} = sdo_util.from_geojson(
:in_geometry,
NULL,
:srid
)
WHERE {self.id_field} = :in_id
"""
# Bind variable for the SDO_GEOMETRY type
in_geometry = self._get_sdo_from_geojson_geometry(
db.conn, request_data.get("geometry").get("coordinates")[0]
)
# in_geometry = self._get_sdo_from_geojson_geometry(
# db.conn, request_data.get("geometry").get("coordinates")[0]
# )
in_geometry = json.dumps(request_data.get("geometry"))
bind_variables = {
**bind_variables,
"in_id": identifier,
"in_geometry": in_geometry,
"srid": self._get_srid_from_crs(self.storage_crs),
}
# SQL manipulation plugin
tests/data/oracle/init-db/01-create-user.sql
Executable
+5
View File
@@ -0,0 +1,5 @@
CONNECT sys/oracle@XEPDB1 AS SYSDBA;
CREATE USER geo_test IDENTIFIED BY geo_test QUOTA UNLIMITED ON USERS;
GRANT CONNECT, RESOURCE, DBA TO geo_test;
tests/load_oracle_data.py
+13
View File
@@ -23,6 +23,19 @@ CREATE TABLE geo_test.lakes (
cur.execute(sql)
sql = """
CREATE PUBLIC SYNONYM lakes_public_syn FOR geo_test.lakes
"""
cur.execute(sql)
sql = """
CREATE SYNONYM geo_test.lakes_private_syn FOR geo_test.lakes
"""
cur.execute(sql)
sql = """
INSERT INTO lakes (area, volume, name, wiki_link, geometry)
VALUES (NULL, NULL, 'Lake Baikal',
tests/test_filesystem_provider.py
+1 -1
View File
@@ -54,7 +54,7 @@ def test_query(config):
r = p.get_data_path(baseurl, urlpath, dirpath)
assert len(r['links']) == 10
assert len(r['links']) == 11
r = p.get_data_path(baseurl, urlpath, '/poi_portugal')
tests/test_oracle_provider.py
+105 -1
View File
@@ -48,9 +48,19 @@ class SqlManipulator:
sql_query,
bind_variables,
sql_manipulator_options,
offset,
limit,
resulttype,
bbox,
source_crs,
datetime_,
properties,
sortby,
skip_geometry,
select_properties,
crs_transform_spec,
q,
language,
filterq,
):
sql = "ID = 10 AND :foo != :bar"
@@ -137,6 +147,44 @@ def config():
}
@pytest.fixture()
def config_public_synonym():
return {
"name": "Oracle",
"type": "feature",
"data": {
"host": HOST,
"port": PORT,
"service_name": SERVICE_NAME,
"user": USERNAME,
"password": PASSWORD,
},
"id_field": "id",
"table": "lakes_public_syn",
"geom_field": "geometry",
"editable": True,
}
@pytest.fixture()
def config_private_synonym():
return {
"name": "Oracle",
"type": "feature",
"data": {
"host": HOST,
"port": PORT,
"service_name": SERVICE_NAME,
"user": USERNAME,
"password": PASSWORD,
},
"id_field": "id",
"table": "lakes_private_syn",
"geom_field": "geometry",
"editable": True,
}
@pytest.fixture()
def config_manipulator(config):
return {
@@ -178,6 +226,18 @@ def create_geojson():
}
@pytest.fixture()
def create_point_geojson():
return {
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [9.603316032965449, 47.48872063967191],
},
"properties": {"name": "Yachthafen Fischerinsel", "wiki_link": None},
}
@pytest.fixture()
def update_geojson():
return {
@@ -235,6 +295,38 @@ def test_get_fields(config):
assert provider.fields == expected_fields
def test_get_fields_private_synonym(config_private_synonym):
"""Test get_fields from private synonym"""
expected_fields = {
"id": {"type": "NUMBER"},
"area": {"type": "NUMBER"},
"volume": {"type": "NUMBER"},
"name": {"type": "VARCHAR2"},
"wiki_link": {"type": "VARCHAR2"},
}
provider = OracleProvider(config_private_synonym)
assert provider.get_fields() == expected_fields
assert provider.fields == expected_fields
def test_get_fields_public_synonym(config_public_synonym):
"""Test get_fields from public synonym"""
expected_fields = {
"id": {"type": "NUMBER"},
"area": {"type": "NUMBER"},
"volume": {"type": "NUMBER"},
"name": {"type": "VARCHAR2"},
"wiki_link": {"type": "VARCHAR2"},
}
provider = OracleProvider(config_public_synonym)
assert provider.get_fields() == expected_fields
assert provider.fields == expected_fields
def test_get_fields_properties(config_properties):
"""
Test get_fields with subset of columns.
@@ -453,3 +545,15 @@ def test_delete_sql_manipulator(config_manipulator, config):
down = p2.query(sortby=[{"property": "id", "order": "-"}])
assert down["features"][0]["id"] == identifier
def test_create_point(config, create_point_geojson):
"""Test simple create"""
p = OracleProvider(config)
result = p.create(create_point_geojson)
assert result == 28
data = p.get(28)
assert data.get("geometry").get("type") == "Point"