mirror of
https://github.com/umap-project/umap.git
synced 2025-04-29 03:42:37 +02:00
Merge pull request #1670 from umap-project/clean-tilelayer-command
chore: add clean_tilelayer management command
This commit is contained in:
Yohan Boniface
GitHub
commit
5836ae859a
2 changed files with 235 additions and 0 deletions
152
umap/management/commands/clean_tilelayer.py
Normal file
152
umap/management/commands/clean_tilelayer.py
Normal file
|
|
@ -0,0 +1,152 @@
|
|||
import json
|
||||
import sys
|
||||
|
||||
from django.core.management.base import BaseCommand
|
||||
from django.db import connection
|
||||
from psycopg.types.json import Jsonb
|
||||
|
||||
from umap.models import Map, TileLayer
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = """Clean tilelayer in database
|
||||
|
||||
This will simply replace the URL in maps settings:
|
||||
umap clean_tilelayer http://my.old/url/template http://my.new/url/template
|
||||
|
||||
This will replace the whole tilelayer in maps settings by the one with this name:
|
||||
umap clean_tilelayer http://my.old/url/template "some string"
|
||||
|
||||
This will replace the whole tilelayer in maps settings by the one with this id:
|
||||
umap clean_tilelayer http://my.old/url/template an_id
|
||||
|
||||
This will delete the whole tilelayer from maps settings:
|
||||
umap clean_tilelayer http://my.old/url/template
|
||||
|
||||
To get the available tilelayers in db (available for users):
|
||||
umap clean_tilelayer --available
|
||||
|
||||
To get statistics of tilelayers usage in db (including custom ones):
|
||||
umap clean_tilelayer --available
|
||||
"""
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument("old", nargs="?", help="url template we want to clean")
|
||||
parser.add_argument(
|
||||
"new", help="what to replace this tilelayer with", nargs="?"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-input", action="store_true", help="Do not ask for confirm."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--available", action="store_true", help="List known tilelayers."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--stats", action="store_true", help="Display stats on tilelayer usage."
|
||||
)
|
||||
|
||||
def handle(self, *args, **options):
|
||||
self.no_input = options["no_input"]
|
||||
if options["available"]:
|
||||
self.list_available()
|
||||
sys.exit()
|
||||
if options["stats"]:
|
||||
self.stats()
|
||||
sys.exit()
|
||||
old = options["old"]
|
||||
new = options["new"]
|
||||
if not old:
|
||||
sys.exit("⚠ You must define an url_template")
|
||||
|
||||
count = Map.objects.filter(
|
||||
settings__properties__tilelayer__url_template=old
|
||||
).count()
|
||||
if not count:
|
||||
self.stdout.write("⚠ No map found. Exiting.")
|
||||
sys.exit()
|
||||
self.stdout.write(f"{count} maps found.")
|
||||
if not new:
|
||||
self.delete(old)
|
||||
elif new.startswith("http"):
|
||||
self.replace_url(old, new)
|
||||
else:
|
||||
# Let's consider it's a name or an id
|
||||
self.replace_tilelayer(old, new)
|
||||
|
||||
def confirm(self, message):
|
||||
if self.no_input:
|
||||
return True
|
||||
result = input("%s (y/N) " % message) or "n"
|
||||
if not result[0].lower() == "y":
|
||||
self.stdout.write("⚠ Action cancelled.")
|
||||
sys.exit()
|
||||
return True
|
||||
|
||||
def delete(self, old):
|
||||
if self.confirm(
|
||||
"Are you sure you want to delete the tilelayer key from all those "
|
||||
"maps settings ?"
|
||||
):
|
||||
with connection.cursor() as cursor:
|
||||
ret = cursor.execute(
|
||||
"UPDATE umap_map "
|
||||
"SET settings['properties'] = (settings->'properties') - 'tilelayer'"
|
||||
"WHERE settings->'properties'->'tilelayer'->'url_template' = %s",
|
||||
[Jsonb(old)],
|
||||
)
|
||||
self.stdout.write(f"✔ Deleted {old} from {ret.rowcount} maps.")
|
||||
|
||||
def replace_url(self, old, new):
|
||||
if self.confirm(
|
||||
f"Are you sure you want to replace '{old}'' by '{new}'' from all those "
|
||||
"map settings ?"
|
||||
):
|
||||
with connection.cursor() as cursor:
|
||||
ret = cursor.execute(
|
||||
"UPDATE umap_map "
|
||||
"SET settings['properties']['tilelayer']['url_template'] = %s "
|
||||
"WHERE settings->'properties'->'tilelayer'->'url_template' = %s",
|
||||
[Jsonb(new), Jsonb(old)],
|
||||
)
|
||||
self.stdout.write(f"✔ Replaced {old} by {new} in {ret.rowcount} maps.")
|
||||
|
||||
def replace_tilelayer(self, old, new):
|
||||
try:
|
||||
tilelayer = TileLayer.objects.get(name=new)
|
||||
except TileLayer.DoesNotExist:
|
||||
try:
|
||||
tilelayer = TileLayer.objects.get(id=new)
|
||||
except (TileLayer.DoesNotExist, ValueError):
|
||||
sys.exit(f"⚠ Cannot find a TileLayer with name or id = '{new}'.")
|
||||
if self.confirm(
|
||||
f"Are you sure you want to replace {old} by '{tilelayer.name}' "
|
||||
"from all those map settings ?"
|
||||
):
|
||||
with connection.cursor() as cursor:
|
||||
ret = cursor.execute(
|
||||
"UPDATE umap_map "
|
||||
"SET settings['properties']['tilelayer'] = %s "
|
||||
"WHERE settings->'properties'->'tilelayer'->'url_template' = %s",
|
||||
[Jsonb(tilelayer.json), Jsonb(old)],
|
||||
)
|
||||
self.stdout.write(
|
||||
f"✔ Replaced {old} by {tilelayer.name} in {ret.rowcount} maps."
|
||||
)
|
||||
|
||||
def list_available(self):
|
||||
tilelayers = TileLayer.objects.all()
|
||||
for tilelayer in tilelayers:
|
||||
print(f"{tilelayer.pk} '{tilelayer.name}' {tilelayer.url_template}")
|
||||
|
||||
def stats(self):
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT COUNT(*) as count, "
|
||||
"settings->'properties'->'tilelayer'->'url_template' as url "
|
||||
"FROM umap_map "
|
||||
"GROUP BY settings->'properties'->'tilelayer'->'url_template' "
|
||||
"ORDER BY count DESC"
|
||||
)
|
||||
res = cursor.fetchall()
|
||||
for count, url in res:
|
||||
print(f"{count}\t{url}")
|
||||
83
umap/tests/test_clean_tilelayer.py
Normal file
83
umap/tests/test_clean_tilelayer.py
Normal file
|
|
@ -0,0 +1,83 @@
|
|||
import pytest
|
||||
from django.core.management import call_command
|
||||
|
||||
from umap.models import Map
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def test_can_delete_tilelayer_from_settings(map):
|
||||
map.settings["properties"]["tilelayer"] = {
|
||||
"name": "My TileLayer",
|
||||
"maxZoom": 18,
|
||||
"minZoom": 0,
|
||||
"attribution": "My attribution",
|
||||
"url_template": "http://{s}.foo.bar.baz/{z}/{x}/{y}.png",
|
||||
}
|
||||
map.save()
|
||||
# Make sure its saved
|
||||
map = Map.objects.get(pk=map.pk)
|
||||
assert "tilelayer" in map.settings["properties"]
|
||||
call_command(
|
||||
"clean_tilelayer", "http://{s}.foo.bar.baz/{z}/{x}/{y}.png", "--no-input"
|
||||
)
|
||||
map = Map.objects.get(pk=map.pk)
|
||||
assert "tilelayer" not in map.settings["properties"]
|
||||
|
||||
|
||||
def test_can_replace_tilelayer_url_in_map_settings(map):
|
||||
map.settings["properties"]["tilelayer"] = {
|
||||
"name": "My TileLayer",
|
||||
"maxZoom": 18,
|
||||
"minZoom": 0,
|
||||
"attribution": "My attribution",
|
||||
"url_template": "http://{s}.foo.bar.baz/{z}/{x}/{y}.png",
|
||||
}
|
||||
map.save()
|
||||
new = "https://{s}.foo.bar.baz/{z}/{x}/{y}.png"
|
||||
call_command(
|
||||
"clean_tilelayer",
|
||||
"http://{s}.foo.bar.baz/{z}/{x}/{y}.png",
|
||||
new,
|
||||
"--no-input",
|
||||
)
|
||||
map = Map.objects.get(pk=map.pk)
|
||||
assert map.settings["properties"]["tilelayer"]["url_template"] == new
|
||||
|
||||
|
||||
def test_can_replace_tilelayer_by_name_in_map_settings(map, tilelayer):
|
||||
map.settings["properties"]["tilelayer"] = {
|
||||
"name": "My TileLayer",
|
||||
"maxZoom": 18,
|
||||
"minZoom": 0,
|
||||
"attribution": "My attribution",
|
||||
"url_template": "http://{s}.foo.bar.baz/{z}/{x}/{y}.png",
|
||||
}
|
||||
map.save()
|
||||
call_command(
|
||||
"clean_tilelayer",
|
||||
"http://{s}.foo.bar.baz/{z}/{x}/{y}.png",
|
||||
tilelayer.name,
|
||||
"--no-input",
|
||||
)
|
||||
map = Map.objects.get(pk=map.pk)
|
||||
assert map.settings["properties"]["tilelayer"] == tilelayer.json
|
||||
|
||||
|
||||
def test_can_replace_tilelayer_by_id_in_map_settings(map, tilelayer):
|
||||
map.settings["properties"]["tilelayer"] = {
|
||||
"name": "My TileLayer",
|
||||
"maxZoom": 18,
|
||||
"minZoom": 0,
|
||||
"attribution": "My attribution",
|
||||
"url_template": "http://{s}.foo.bar.baz/{z}/{x}/{y}.png",
|
||||
}
|
||||
map.save()
|
||||
call_command(
|
||||
"clean_tilelayer",
|
||||
"http://{s}.foo.bar.baz/{z}/{x}/{y}.png",
|
||||
tilelayer.pk,
|
||||
"--no-input",
|
||||
)
|
||||
map = Map.objects.get(pk=map.pk)
|
||||
assert map.settings["properties"]["tilelayer"] == tilelayer.json
|
||||
Loading…
Reference in a new issue