Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# total primary key length has an upper limit in MySQL
key_size = self._MAX_PRIMARY_KEY_LENGTH // len(keys)
key_type = f'VARCHAR({key_size})'
with pymysql.connect(host=self._db_args.host, user=self._db_args.user,
password=self._db_args.password, port=self._db_args.port,
read_timeout=self.DB_CONNECTION_TIMEOUT,
write_timeout=self.DB_CONNECTION_TIMEOUT,
binary_prefix=True, charset='utf8mb4') as con:
con.execute(f'CREATE DATABASE {self._db_args.db}')
with self._connect(check=False):
cursor = self._cursor
cursor.execute(f'CREATE TABLE terracotta (version VARCHAR(255)) '
f'CHARACTER SET {self._CHARSET}')
cursor.execute('INSERT INTO terracotta VALUES (%s)', [str(__version__)])
cursor.execute(f'CREATE TABLE key_names (key_name {key_type}, '
f'description VARCHAR(8000)) CHARACTER SET {self._CHARSET}')
key_rows = [(key, key_descriptions[key]) for key in keys]
cursor.executemany('INSERT INTO key_names VALUES (%s, %s)', key_rows)
key_string = ', '.join([f'{key} {key_type}' for key in keys])
cursor.execute(f'CREATE TABLE datasets ({key_string}, filepath VARCHAR(8000), '
f'PRIMARY KEY({", ".join(keys)})) CHARACTER SET {self._CHARSET}')
column_string = ', '.join(f'{col} {col_type}' for col, col_type
in self._METADATA_COLUMNS)
cursor.execute(f'CREATE TABLE metadata ({key_string}, {column_string}, '
f'PRIMARY KEY ({", ".join(keys)})) CHARACTER SET {self._CHARSET}')
# invalidate key cache
def test_version_match(driver_path, provider):
from terracotta import drivers, __version__
db = drivers.get_driver(driver_path, provider=provider)
keys = ('some', 'keynames')
db.create(keys)
assert __version__ == db.db_version
def test_get_spec(client):
from terracotta import __version__
rv = client.get('/swagger.json')
assert rv.status_code == 200
assert json.loads(rv.data)
assert __version__ in rv.data.decode('utf-8')
rv = client.get('/apidoc')
assert rv.status_code == 200
assert b'Terracotta' in rv.data
def test_version_match(tmpdir):
from terracotta import drivers, __version__
dbfile = tmpdir.join('test.sqlite')
db = drivers.get_driver(str(dbfile), provider='sqlite')
keys = ('some', 'keys')
db.create(keys)
assert __version__ == db.db_version
'that Terracotta is running on the server', err=True
)
raise click.Abort()
# catch version incompatibility
spec_url = f'{terracotta_hostname}/swagger.json'
with urllib.request.urlopen(spec_url, timeout=5) as response:
spec = json.loads(response.read())
def versiontuple(version_string: str) -> Sequence[str]:
return version_string.split('.')
remote_version = spec['info']['version']
if versiontuple(__version__)[:2] != versiontuple(remote_version)[:2]:
click.echo(
f'Incompatible version: The server running at {terracotta_hostname} is '
f'using Terracotta v{remote_version}, but this is v{__version__}',
err=True
)
raise click.Abort()
# find suitable port
port_range = [port] if port is not None else range(5100, 5200)
port = find_open_port(port_range)
if port is None:
click.echo(f'Could not find open port to bind to (ports tried: {port_range})', err=True)
raise click.Abort()
def open_browser() -> None:
webbrowser.open(f'http://127.0.0.1:{port}/')
if not no_browser:
def _connection_callback(self) -> None:
"""Called after opening a new connection"""
# check for version compatibility
def versiontuple(version_string: str) -> Sequence[str]:
return version_string.split('.')
db_version = self.db_version
current_version = __version__
if versiontuple(db_version)[:2] != versiontuple(current_version)[:2]:
raise exceptions.InvalidDatabaseError(
f'Version conflict: database was created in v{db_version}, '
f'but this is v{current_version}'
def _connection_callback(self) -> None:
if not self._version_checked:
# check for version compatibility
def versiontuple(version_string: str) -> Sequence[str]:
return version_string.split('.')
db_version = self.db_version
current_version = __version__
if versiontuple(db_version)[:2] != versiontuple(current_version)[:2]:
raise exceptions.InvalidDatabaseError(
f'Version conflict: database was created in v{db_version}, '
f'but this is v{current_version}'
)
self._version_checked = True
@click.version_option(version=__version__)
@click.pass_context
def cli(ctx: click.Context,
config: Mapping[str, Any] = None,
loglevel: str = None) -> None:
"""The command line interface for the Terracotta tile server.
All flags must be passed before specifying a subcommand.
Example:
$ terracotta -c config.toml connect localhost:5000
"""
if ctx.invoked_subcommand is None:
click.echo(ctx.get_help())
#
# import os
# import sys
# sys.path.insert(0, os.path.abspath('.'))
# -- Project information -----------------------------------------------------
import re
from terracotta import __version__
project = 'Terracotta'
copyright = '2018-2020, the Terracotta contributors'
author = 'Dion Häfner, Philip Graae'
# The short X.Y version
version = re.match(r'(\d+\.\d+\.\d+)', __version__).group(1)
# The full version, including alpha/beta/rc tags
release = __version__
# -- General configuration ---------------------------------------------------
# If your documentation needs a minimal Sphinx version, state it here.
#
# needs_sphinx = '1.0'
# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = [
'sphinx.ext.intersphinx',
'sphinx.ext.autodoc',
import marshmallow
from terracotta import exceptions, __version__
# define blueprints, will be populated by submodules
tile_api = Blueprint('tile_api', 'terracotta.server')
metadata_api = Blueprint('metadata_api', 'terracotta.server')
spec_api = Blueprint('spec_api', 'terracotta.server')
CORS(metadata_api) # allow access to metadata from all sources
# create an APISpec
spec = APISpec(
title='Terracotta',
version=__version__,
openapi_version='2.0',
info=dict(
description='A modern XYZ Tile Server in Python'
),
plugins=[
FlaskPlugin(),
MarshmallowPlugin()
],
)
def abort(status_code: int, message: str = '') -> Any:
response = jsonify({'message': message})
response.status_code = status_code
return response