Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add data class for workspace WMS settings #48

Merged
merged 1 commit into from
Nov 15, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 85 additions & 4 deletions geoservercloud/geoservercloud.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from geoservercloud.models.featuretype import FeatureType
from geoservercloud.models.layer import Layer
from geoservercloud.models.style import Style
from geoservercloud.models.wmssettings import WmsSettings
from geoservercloud.models.workspace import Workspace
from geoservercloud.services import OwsService, RestService
from geoservercloud.templates import Templates
Expand Down Expand Up @@ -120,20 +121,100 @@ def recreate_workspace(
set_default_workspace=set_default_workspace,
)

def publish_workspace(self, workspace_name) -> tuple[str, int]:
def get_workspace_wms_settings(
self, workspace_name: str
) -> tuple[dict[str, Any] | str, int]:
"""
Get the WMS settings for a given workspace
"""
wms_settings, status_code = self.rest_service.get_workspace_wms_settings(
workspace_name
)
if isinstance(wms_settings, str):
return wms_settings, status_code
return wms_settings.asdict(), status_code

def publish_workspace(
self,
workspace_name: str,
versions: list[str] = ["1.1.1", "1.3.0"],
cite_compliant: bool = False,
schema_base_url: str = "http://schemas.opengis.net",
verbose: bool = False,
bbox_for_each_crs: bool = False,
watermark: dict = {
"enabled": False,
"position": "BOT_RIGHT",
"transparency": 100,
},
interpolation: str = "Nearest",
get_feature_info_mime_type_checking_enabled: bool = False,
get_map_mime_type_checking_enabled: bool = False,
dynamic_styling_disabled: bool = False,
features_reprojection_disabled: bool = False,
max_buffer: int = 0,
max_request_memory: int = 0,
max_rendering_time: int = 0,
max_rendering_errors: int = 0,
max_requested_dimension_values: int = 100,
cache_configuration: dict = {
"enabled": False,
"maxEntries": 1000,
"maxEntrySize": 51200,
},
remote_style_max_request_time: int = 60000,
remote_style_timeout: int = 30000,
default_group_style_enabled: bool = True,
transform_feature_info_disabled: bool = False,
auto_escape_template_values: bool = False,
) -> tuple[str, int]:
"""
Publish the WMS service for a given workspace
"""
return self.rest_service.publish_workspace(Workspace(workspace_name))
wms_settings = WmsSettings(
workspace_name=workspace_name,
name="WMS",
enabled=True,
versions={
"org.geotools.util.Version": [
{"version": version} for version in versions
]
},
cite_compliant=cite_compliant,
schema_base_url=schema_base_url,
verbose=verbose,
bbox_for_each_crs=bbox_for_each_crs,
watermark=watermark,
interpolation=interpolation,
get_feature_info_mime_type_checking_enabled=get_feature_info_mime_type_checking_enabled,
get_map_mime_type_checking_enabled=get_map_mime_type_checking_enabled,
dynamic_styling_disabled=dynamic_styling_disabled,
features_reprojection_disabled=features_reprojection_disabled,
max_buffer=max_buffer,
max_request_memory=max_request_memory,
max_rendering_time=max_rendering_time,
max_rendering_errors=max_rendering_errors,
max_requested_dimension_values=max_requested_dimension_values,
cache_configuration=cache_configuration,
remote_style_max_request_time=remote_style_max_request_time,
remote_style_timeout=remote_style_timeout,
default_group_style_enabled=default_group_style_enabled,
transform_feature_info_disabled=transform_feature_info_disabled,
auto_escape_template_values=auto_escape_template_values,
)
return self.rest_service.put_workspace_wms_settings(
workspace_name, wms_settings
)

def set_default_locale_for_service(
self, workspace_name: str, locale: str | None
) -> tuple[str, int]:
"""
Set a default language for localized WMS requests
"""
return self.rest_service.set_default_locale_for_service(
Workspace(workspace_name), locale
wms_settings = WmsSettings(default_locale=locale)
return self.rest_service.put_workspace_wms_settings(
workspace_name, wms_settings
)

def unset_default_locale_for_service(self, workspace_name) -> tuple[str, int]:
Expand Down
159 changes: 159 additions & 0 deletions geoservercloud/models/wmssettings.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
from typing import Any

from geoservercloud.models.common import EntityModel, ReferencedObjectModel


class WmsSettings(EntityModel):
def __init__(
self,
workspace_name: str | None = None,
name: str | None = None,
enabled: bool | None = None,
default_locale: str | None = None,
auto_escape_template_values: bool | None = None,
bbox_for_each_crs: bool | None = None,
cache_configuration: dict[str, bool | int] | None = None,
cite_compliant: bool | None = None,
default_group_style_enabled: bool | None = None,
dynamic_styling_disabled: bool | None = None,
features_reprojection_disabled: bool | None = None,
get_feature_info_mime_type_checking_enabled: bool | None = None,
get_map_mime_type_checking_enabled: bool | None = None,
interpolation: str | None = None,
max_buffer: int | None = None,
max_rendering_errors: int | None = None,
max_rendering_time: int | None = None,
max_request_memory: int | None = None,
max_requested_dimension_values: int | None = None,
remote_style_max_request_time: int | None = None,
remote_style_timeout: int | None = None,
schema_base_url: str | None = None,
transform_feature_info_disabled: bool | None = None,
verbose: bool | None = None,
versions: dict[str, list[dict[str, str]]] | None = None,
watermark: dict[str, bool | int | str] | None = None,
) -> None:
self.workspace: ReferencedObjectModel | None = (
ReferencedObjectModel(workspace_name) if workspace_name else None
)
self.name: str | None = name
self.enabled: bool | None = enabled
self.default_locale: str | None = default_locale
self.auto_escape_template_values: bool | None = auto_escape_template_values
self.bbox_for_each_crs: bool | None = bbox_for_each_crs
self.cache_configuration: dict[str, bool | int] | None = cache_configuration
self.cite_compliant: bool | None = cite_compliant
self.default_group_style_enabled: bool | None = default_group_style_enabled
self.dynamic_styling_disabled: bool | None = dynamic_styling_disabled
self.features_reprojection_disabled: bool | None = (
features_reprojection_disabled
)
self.get_feature_info_mime_type_checking_enabled: bool | None = (
get_feature_info_mime_type_checking_enabled
)
self.get_map_mime_type_checking_enabled: bool | None = (
get_map_mime_type_checking_enabled
)
self.interpolation: str | None = interpolation
self.max_buffer: int | None = max_buffer
self.max_rendering_errors: int | None = max_rendering_errors
self.max_rendering_time: int | None = max_rendering_time
self.max_request_memory: int | None = max_request_memory
self.max_requested_dimension_values: int | None = max_requested_dimension_values
self.remote_style_max_request_time: int | None = remote_style_max_request_time
self.remote_style_timeout: int | None = remote_style_timeout
self.schema_base_url: str | None = schema_base_url
self.transform_feature_info_disabled: bool | None = (
transform_feature_info_disabled
)
self.verbose: bool | None = verbose
self.versions: dict[str, list[dict[str, str]]] | None = versions
self.watermark: dict[str, bool | int | str] | None = watermark

@property
def workspace_name(self) -> str | None:
return self.workspace.name if self.workspace else None

def asdict(self) -> dict[str, str]:
return EntityModel.add_items_to_dict(
{},
{
"workspace": self.workspace_name,
"name": self.name,
"enabled": self.enabled,
"defaultLocale": self.default_locale,
"autoEscapeTemplateValues": self.auto_escape_template_values,
"bboxForEachCRS": self.bbox_for_each_crs,
"cacheConfiguration": self.cache_configuration,
"citeCompliant": self.cite_compliant,
"defaultGroupStyleEnabled": self.default_group_style_enabled,
"dynamicStylingDisabled": self.dynamic_styling_disabled,
"featuresReprojectionDisabled": self.features_reprojection_disabled,
"getFeatureInfoMimeTypeCheckingEnabled": self.get_feature_info_mime_type_checking_enabled,
"getMapMimeTypeCheckingEnabled": self.get_map_mime_type_checking_enabled,
"interpolation": self.interpolation,
"maxBuffer": self.max_buffer,
"maxRenderingErrors": self.max_rendering_errors,
"maxRenderingTime": self.max_rendering_time,
"maxRequestMemory": self.max_request_memory,
"maxRequestedDimensionValues": self.max_requested_dimension_values,
"remoteStyleMaxRequestTime": self.remote_style_max_request_time,
"remoteStyleTimeout": self.remote_style_timeout,
"schemaBaseURL": self.schema_base_url,
"transformFeatureInfoDisabled": self.transform_feature_info_disabled,
"verbose": self.verbose,
"versions": self.versions,
"watermark": self.watermark,
},
)

def post_payload(self) -> dict[str, Any]:
content: dict[str, Any] = self.asdict()
if self.workspace:
content["workspace"] = self.workspace.asdict()
return {"wms": content}

def put_payload(self) -> dict[str, Any]:
return self.post_payload()

@classmethod
def from_get_response_payload(cls, content: dict[str, Any]):
wms_settings = content["wms"]
return cls(
workspace_name=wms_settings.get("workspace", {}).get("name"),
name=wms_settings.get("name"),
enabled=wms_settings.get("enabled"),
default_locale=wms_settings.get("defaultLocale"),
auto_escape_template_values=wms_settings.get("autoEscapeTemplateValues"),
bbox_for_each_crs=wms_settings.get("bboxForEachCRS"),
cache_configuration=wms_settings.get("cacheConfiguration"),
cite_compliant=wms_settings.get("citeCompliant"),
default_group_style_enabled=wms_settings.get("defaultGroupStyleEnabled"),
dynamic_styling_disabled=wms_settings.get("dynamicStylingDisabled"),
features_reprojection_disabled=wms_settings.get(
"featuresReprojectionDisabled"
),
get_feature_info_mime_type_checking_enabled=wms_settings.get(
"getFeatureInfoMimeTypeCheckingEnabled"
),
get_map_mime_type_checking_enabled=wms_settings.get(
"getMapMimeTypeCheckingEnabled"
),
interpolation=wms_settings.get("interpolation"),
max_buffer=wms_settings.get("maxBuffer"),
max_rendering_errors=wms_settings.get("maxRenderingErrors"),
max_rendering_time=wms_settings.get("maxRenderingTime"),
max_request_memory=wms_settings.get("maxRequestMemory"),
max_requested_dimension_values=wms_settings.get(
"maxRequestedDimensionValues"
),
remote_style_max_request_time=wms_settings.get("remoteStyleMaxRequestTime"),
remote_style_timeout=wms_settings.get("remoteStyleTimeout"),
schema_base_url=wms_settings.get("schemaBaseURL"),
transform_feature_info_disabled=wms_settings.get(
"transformFeatureInfoDisabled"
),
verbose=wms_settings.get("verbose"),
versions=wms_settings.get("versions"),
watermark=wms_settings.get("watermark"),
)
24 changes: 11 additions & 13 deletions geoservercloud/services/restservice.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from geoservercloud.models.resourcedirectory import ResourceDirectory
from geoservercloud.models.style import Style
from geoservercloud.models.styles import Styles
from geoservercloud.models.wmssettings import WmsSettings
from geoservercloud.models.workspace import Workspace
from geoservercloud.models.workspaces import Workspaces
from geoservercloud.services.restclient import RestClient
Expand Down Expand Up @@ -62,23 +63,20 @@ def delete_workspace(self, workspace: Workspace) -> tuple[str, int]:
response: Response = self.rest_client.delete(path, params=params)
return response.content.decode(), response.status_code

def publish_workspace(self, workspace: Workspace) -> tuple[str, int]:
data: dict[str, dict[str, Any]] = Templates.workspace_wms(workspace.name)
response: Response = self.rest_client.put(
self.rest_endpoints.workspace_wms_settings(workspace.name), json=data
def get_workspace_wms_settings(
self, workspace_name: str
) -> tuple[WmsSettings | str, int]:
response: Response = self.rest_client.get(
self.rest_endpoints.workspace_wms_settings(workspace_name)
)
return response.content.decode(), response.status_code
return self.deserialize_response(response, WmsSettings)

def set_default_locale_for_service(
self, workspace: Workspace, locale: str | None
def put_workspace_wms_settings(
self, workspace_name: str, wms_settings: WmsSettings
) -> tuple[str, int]:
data: dict[str, dict[str, Any]] = {
"wms": {
"defaultLocale": locale,
}
}
response: Response = self.rest_client.put(
self.rest_endpoints.workspace_wms_settings(workspace.name), json=data
self.rest_endpoints.workspace_wms_settings(workspace_name),
json=wms_settings.put_payload(),
)
return response.content.decode(), response.status_code

Expand Down
45 changes: 0 additions & 45 deletions geoservercloud/templates.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,51 +4,6 @@


class Templates:
@staticmethod
def workspace_wms(workspace: str) -> dict[str, dict[str, Any]]:
return {
"wms": {
"workspace": {"name": workspace},
"enabled": True,
"name": "WMS",
"versions": {
"org.geotools.util.Version": [
{"version": "1.1.1"},
{"version": "1.3.0"},
]
},
"citeCompliant": False,
"schemaBaseURL": "http://schemas.opengis.net",
"verbose": False,
"bboxForEachCRS": False,
"watermark": {
"enabled": False,
"position": "BOT_RIGHT",
"transparency": 100,
},
"interpolation": "Nearest",
"getFeatureInfoMimeTypeCheckingEnabled": False,
"getMapMimeTypeCheckingEnabled": False,
"dynamicStylingDisabled": False,
"featuresReprojectionDisabled": False,
"maxBuffer": 0,
"maxRequestMemory": 0,
"maxRenderingTime": 0,
"maxRenderingErrors": 0,
"maxRequestedDimensionValues": 100,
"cacheConfiguration": {
"enabled": False,
"maxEntries": 1000,
"maxEntrySize": 51200,
},
"remoteStyleMaxRequestTime": 60000,
"remoteStyleTimeout": 30000,
"defaultGroupStyleEnabled": True,
"transformFeatureInfoDisabled": False,
"autoEscapeTemplateValues": False,
}
}

@staticmethod
def wmts_store(
workspace: str, name: str, capabilities: str
Expand Down
Loading