Source code for operators.tile.remove_tile
from typing import Literal, List, Union
from airflow.utils.decorators import apply_defaults
from airflow.utils.context import Context
from airflow.exceptions import AirflowException
from operators.tile.base_tile import BaseTileOperator
[docs]
class RemoveTileOperator(BaseTileOperator):
"""
Operator to remove tile(s) from the tileserver without archiving to S3.
:param src_env: source environment, defaults to "main"
:type src_env: Literal["main", "staging"]
:param src_tiles: single path or list of paths to the tilesets (starting at home directory `~/`)
:type src_tiles: Union[str, List[str]]
:param is_dry_run: whether to perform a dry run, defaults to True
:type is_dry_run: bool, optional
:param verify_tile_exists: whether to verify if the tile exists before performing the action, defaults to True
:type verify_tile_exists: bool, optional
:raises AirflowException: if the removal fails
"""
ui_color = "#ff9999" # Light red shade
ui_fgcolor = "#000000"
custom_operator_name = "RemoveTileOperator"
template_fields = ["src_tiles", "src_env", "is_dry_run"]
@apply_defaults
def __init__(
self,
src_tiles: Union[str, List[str]],
src_env: Literal["main", "staging"] = "main",
is_dry_run: bool = True,
verify_tile_exists: bool = True,
*args,
**kwargs,
):
super().__init__(
src_env=src_env,
src_tiles=src_tiles,
is_dry_run=is_dry_run,
verify_tile_exists=verify_tile_exists,
*args,
**kwargs,
)
[docs]
def execute(self, context: Context) -> None:
"""
Remove tile(s) from the tileserver without archiving to S3.
"""
try:
# Add clear indication of dry run mode at the beginning
if self.is_dry_run:
self.log.info(
f"╔═════════════════════════════════════════════════════════════════╗"
)
self.log.info(
f"║ 🔍 DRY RUN MODE: No actual changes will be made to files or systems ║"
)
self.log.info(
f"╚═════════════════════════════════════════════════════════════════╝"
)
self.log.info(
f"Starting RemoveTileOperator with {len(self.src_tiles)} tiles in environment {self.env_folder}"
)
# Verify tiles exist before proceeding
environment_alias = self.tileserver_alias()
if not self.is_dry_run and self.verify_tile_exists:
for src_tile in self.src_tiles:
self.verify_mbtiles_file(src_tile, environment_alias)
# List of tileset names for logging
tileset_names = [self.get_tileset_name(path) for path in self.src_tiles]
self.log.info(
f"Removing tiles from tileserver {environment_alias}: {tileset_names}"
)
if self.is_dry_run:
# Dry run path - just log what would happen
self.log.info(
f"[DRY RUN] Would remove {len(self.src_tiles)} tiles from server: {environment_alias}"
)
for src_tile in self.src_tiles:
self.log.info(
f"[DRY RUN] Would remove file from server: {environment_alias}:{src_tile}"
)
self.log.info(f"[DRY RUN] Would stop tileserver on {environment_alias}")
self.log.info(
f"[DRY RUN] Would update tileserver config, then restart tileserver {environment_alias}"
)
self.log.info(
f"[DRY RUN] Would remove tiles: {', '.join(tileset_names)}"
)
else:
# Actual execution path
self.log.info(
f"Removing {len(self.src_tiles)} tiles from server: {environment_alias}"
)
try:
# Remove each file
for src_tile in self.src_tiles:
tileset_name = self.get_tileset_name(src_tile)
self.log.info(
f"Removing tile {tileset_name} from {environment_alias}:{src_tile}"
)
self.execute_ssh_command(
environment_alias,
f"rm -f {src_tile}",
f"remove file from server: {src_tile}",
)
# Restart tileserver ONCE after all files are processed
self.restart_tileserver(environment_alias)
self.log.info(
f"Successfully removed {len(self.src_tiles)} tiles from server"
)
except Exception as e:
# Attempt to restart the tileserver in case of failure
self.log.error(f"Error during remove operation: {e}")
try:
self.restart_tileserver(environment_alias)
except Exception as restart_error:
self.log.error(
f"Failed to restart tileserver after error: {restart_error}"
)
raise AirflowException(f"Failed to remove tiles from server: {e}")
# Remove from datahub after server operations
for tileset_name in tileset_names:
try:
self.remove_from_datahub(tileset_name)
except Exception as e:
self.log.error(
f"Error removing tileset {tileset_name} from datahub: {e}"
)
# Add clear indication of completion
if self.is_dry_run:
self.log.info(
f"╔════════════════════════════════════════════════════════════════╗"
)
self.log.info(
f"║ ✓ DRY RUN COMPLETED: No actual changes were made ║"
)
self.log.info(
f"╚════════════════════════════════════════════════════════════════╝"
)
else:
self.log.info(
f"✓ Removal successfully completed for {len(self.src_tiles)} tiles"
)
except Exception as ex:
self.log.error(f"Error in RemoveTileOperator: {ex}")
# Send notification to Slack for failures
error_message = f"""
⚠️ Error in RemoveTileOperator:
Environment: {self.env_folder}
Tiles Count: {len(self.src_tiles)}
{'[DRY RUN] - No actual changes were made' if self.is_dry_run else ''}
Error: {str(ex)}
"""
self.notify_slack(error_message)
raise ex