Coverage for pygeoapi/util.py: 68%

1# =================================================================

3# Authors: Tom Kralidis <tomkralidis@gmail.com>

7# Permission is hereby granted, free of charge, to any person

8# obtaining a copy of this software and associated documentation

9# files (the "Software"), to deal in the Software without

10# restriction, including without limitation the rights to use,

11# copy, modify, merge, publish, distribute, sublicense, and/or sell

12# copies of the Software, and to permit persons to whom the

13# Software is furnished to do so, subject to the following

14# conditions:

15#

16# The above copyright notice and this permission notice shall be

17# included in all copies or substantial portions of the Software.

18#

19# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,

20# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES

21# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND

22# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT

23# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,

24# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING

25# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR

26# OTHER DEALINGS IN THE SOFTWARE.

27#

28# =================================================================

30"""Generic util functions used in the code"""

32import base64

33from filelock import FileLock

34from dataclasses import dataclass

35from datetime import date, datetime, time, timezone

36from decimal import Decimal

37from enum import Enum

38from heapq import heappush

39import ipaddress

40import json

41import logging

42import mimetypes

43import os

44import pathlib

45from pathlib import Path

46import re

47import socket

48from typing import Any, IO, Union, List, Optional

49from urllib.parse import urlparse

50from urllib.request import urlopen

51import uuid

53import dateutil.parser

54from babel.support import Translations

55from jinja2 import Environment, FileSystemLoader, select_autoescape

56from jinja2.exceptions import TemplateNotFound

57from requests import Session

58from requests.structures import CaseInsensitiveDict

59from shapely.geometry import (

60 box,

61 Polygon,

62 mapping as geom_to_geojson,

63)

64import yaml

66from pygeoapi import __version__

67from pygeoapi import l10n

68from pygeoapi.models import config as config_models

69from pygeoapi.plugin import load_plugin, PLUGINS

70from pygeoapi.provider import get_provider_default

72LOGGER = logging.getLogger(__name__)

74DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'

76THISDIR = Path(__file__).parent.resolve()

77RESOURCESDIR = THISDIR / 'resources'

78TEMPLATESDIR = THISDIR / 'templates'

79DEFINITIONSDIR = RESOURCESDIR / 'definitions'

80SCHEMASDIR = RESOURCESDIR / 'schemas'

83mimetypes.add_type('text/plain', '.yaml')

84mimetypes.add_type('text/plain', '.yml')

87def dategetter(date_property: str, collection: dict) -> str:

88 """

89 Attempts to obtain a date value from a collection.

91 :param date_property: property representing the date

92 :param collection: dictionary to check within

94 :returns: `str` (ISO8601) representing the date (allowing

95 for an open interval using null)

96 """

98 value = collection.get(date_property)

100 if value is None or isinstance(value, str):

101 return value

102 else:

103 return value.isoformat()

104

105

106def get_typed_value(value: str) -> Union[bool, float, int, str]:

107 """

108 Derive true type from data value

109

110 :param value: value

111

112 :returns: value as a native Python data type

113 """

114

115 try:

116 if '.' in value: # float?

117 value2 = float(value)

118 elif len(value) > 1 and value.startswith('0'): 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true

119 value2 = value

120 elif value.lower() in ['true', 'false']:

121 value2 = str2bool(value)

122 else: # int?

123 value2 = int(value)

124 except ValueError: # string (default)?

125 value2 = value

126

127 return value2

128

129

130def yaml_load(fh: IO) -> dict:

131 """

132 serializes a YAML files into a pyyaml object

133

134 :param fh: file handle

135

136 :returns: `dict` representation of YAML

137 """

138

139 # # support environment variables in config

140 # # https://stackoverflow.com/a/55301129

141

142 env_matcher = re.compile(

143 r'.*?\$\{(?P<varname>\w+)(:-(?P<default>[^}]*))?\}')

144

145 def env_constructor(loader, node):

146 result = ""

147 current_index = 0

148 raw_value = node.value

149 for match_obj in env_matcher.finditer(raw_value):

150 groups = match_obj.groupdict()

151 varname_start = match_obj.span('varname')[0]

152 result += raw_value[current_index:(varname_start-2)]

153 if (var_value := os.getenv(groups['varname'])) is not None:

154 result += var_value

155 elif (default_value := groups.get('default')) is not None: 155 ↛ 158line 155 didn't jump to line 158 because the condition on line 155 was always true

156 result += default_value

157 else:

158 raise EnvironmentError(

159 f'Could not find the {groups["varname"]!r} environment '

160 f'variable'

161 )

162 current_index = match_obj.end()

163 else:

164 result += raw_value[current_index:]

165 return get_typed_value(result)

166

167 class EnvVarLoader(yaml.SafeLoader):

168 pass

169

170 EnvVarLoader.add_implicit_resolver('!env', env_matcher, None)

171 EnvVarLoader.add_constructor('!env', env_constructor)

172 return yaml.load(fh, Loader=EnvVarLoader)

173

174

175def get_api_rules(config: dict) -> config_models.APIRules:

176 """ Extracts the default API design rules from the given configuration.

177

178 :param config: Current pygeoapi configuration (dictionary).

179 :returns: An APIRules instance.

180 """

181 rules = config['server'].get('api_rules') or {}

182 rules.setdefault('api_version', __version__)

183 return config_models.APIRules.create(**rules)

184

185

186def get_base_url(config: dict) -> str:

187 """ Returns the full pygeoapi base URL. """

188 rules = get_api_rules(config)

189 return url_join(config['server']['url'], rules.get_url_prefix())

190

191

192def yaml_dump(dict_: dict, destfile: str) -> bool:

193 """

194 Dump dict to YAML file

195

196 :param dict_: `dict` to dump

197 :param destfile: destination filepath

198

199 :returns: `bool`

200 """

201

202 def path_representer(dumper, data):

203 return dumper.represent_scalar(u'tag:yaml.org,2002:str', str(data))

204

205 yaml.add_multi_representer(pathlib.PurePath, path_representer)

206

207 def datetime_representer(dumper, data: datetime):

208 if data.tzinfo is None:

209 data = data.replace(tzinfo=timezone.utc)

210 else:

211 data = data.astimezone(timezone.utc)

212 value = data.strftime("%Y-%m-%dT%H:%M:%SZ")

213

214 # timestamp in a specified format, without string quotes

215 return dumper.represent_scalar(u'tag:yaml.org,2002:timestamp', value)

216

217 yaml.add_representer(datetime, datetime_representer)

218

219 lock = FileLock(f'{destfile}.lock')

220

221 with lock:

222 LOGGER.debug('Dumping YAML document')

223 with open(destfile, 'wb') as fh:

224 yaml.dump(dict_, fh, sort_keys=False, encoding='utf8',

225 indent=4, default_flow_style=False,

226 allow_unicode=True)

227

228 return True

229

230

231def str2bool(value: Union[bool, str]) -> bool:

232 """

233 helper function to return Python boolean

234 type (source: https://stackoverflow.com/a/715468)

235

236 :param value: value to be evaluated

237

238 :returns: `bool` of whether the value is boolean-ish

239 """

240

241 value2 = False

242

243 if isinstance(value, bool):

244 value2 = value

245 else:

246 value2 = value.lower() in ('yes', 'true', 't', '1', 'on')

247

248 return value2

249

250

251def to_json(dict_: dict, pretty: bool = False) -> str:

252 """

253 Serialize dict to JSON

254

255 :param dict_: `dict` of JSON representation

256 :param pretty: `bool` of whether to prettify JSON (default is `False`)

257

258 :returns: JSON string representation

259 """

260

261 if pretty:

262 indent = 4

263 else:

264 indent = None

265

266 LOGGER.debug('Dumping JSON')

267 json_dump = json.dumps(dict_, default=json_serial, indent=indent,

268 separators=(',', ':'))

269

270 LOGGER.debug('Removing < and >')

271 json_dump = json_dump.replace('<', '&lt').replace('>', '&gt')

272

273 return json_dump

274

275

276def format_datetime(value: str, format_: str = DATETIME_FORMAT) -> str:

277 """

278 Parse datetime as ISO 8601 string; re-present it in particular format

279 for display in HTML

280

281 :param value: `str` of ISO datetime

282 :param format_: `str` of datetime format for strftime

283

284 :returns: string

285 """

286

287 if not isinstance(value, str) or not value.strip():

288 return ''

289

290 return dateutil.parser.isoparse(value).strftime(format_)

291

292

293def get_current_datetime(tz: timezone = timezone.utc,

294 format_: str = DATETIME_FORMAT) -> str:

295 return datetime.now(tz).strftime(format_)

296

297

298def file_modified_iso8601(filepath: Path) -> str:

299 """

300 Provide a file's ctime in ISO8601

301

302 :param filepath: path to file

303

304 :returns: string of ISO8601

305 """

306

307 return datetime.fromtimestamp(

308 os.path.getctime(filepath)).strftime('%Y-%m-%dT%H:%M:%SZ')

309

310

311def human_size(nbytes: int) -> str:

312 """

313 Provides human readable file size

314

315 source: https://stackoverflow.com/a/14996816

316

317 :param nbytes: int of file size (bytes)

318 :param units: list of unit abbreviations

319

320 :returns: string of human readable filesize

321 """

322

323 suffixes = ['B', 'K', 'M', 'G', 'T', 'P']

324

325 i = 0

326

327 while nbytes >= 1024 and i < len(suffixes)-1:

328 nbytes /= 1024.

329 i += 1

330

331 if suffixes[i] == 'K':

332 f = str(int(nbytes)).rstrip('0').rstrip('.')

333 elif suffixes[i] == 'B':

334 return nbytes

335 else:

336 f = f'{nbytes:.1f}'.rstrip('0').rstrip('.')

337

338 return f'{f}{suffixes[i]}'

339

340

341def format_duration(start: str, end: str = None) -> str:

342 """

343 Parse a start and (optional) end datetime as ISO 8601 strings, calculate

344 the difference, and return that duration as a string.

345

346 :param start: `str` of ISO datetime

347 :param end: `str` of ISO datetime, defaults to `start` for a 0 duration

348

349 :returns: string

350 """

351

352 if not isinstance(start, str) or not start.strip(): 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true

353 return ''

354 end = end or start

355 duration = dateutil.parser.isoparse(end) - dateutil.parser.isoparse(start)

356 return str(duration)

357

358

359def get_path_basename(urlpath: str) -> str:

360 """

361 Helper function to derive file basename

362

363 :param urlpath: URL path

364

365 :returns: string of basename of URL path

366 """

367

368 return Path(urlpath).name

369

370

371def json_serial(obj: Any) -> str:

372 """

373 helper function to convert to JSON non-default

374 types (source: https://stackoverflow.com/a/22238613)

375

376 :param obj: `object` to be evaluated

377

378 :returns: JSON non-default type to `str`

379 """

380

381 if isinstance(obj, (datetime, date, time)): 381 ↛ 382line 381 didn't jump to line 382 because the condition on line 381 was never true

382 return obj.isoformat()

383 elif isinstance(obj, bytes): 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true

384 try:

385 LOGGER.debug('Returning as UTF-8 decoded bytes')

386 return obj.decode('utf-8')

387 except UnicodeDecodeError:

388 LOGGER.debug('Returning as base64 encoded JSON object')

389 return base64.b64encode(obj)

390 elif isinstance(obj, Decimal): 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true

391 return float(obj)

392 elif type(obj).__name__ in ['int32', 'int64']: 392 ↛ 393line 392 didn't jump to line 393 because the condition on line 392 was never true

393 return int(obj)

394 elif type(obj).__name__ in ['float32', 'float64']: 394 ↛ 395line 394 didn't jump to line 395 because the condition on line 394 was never true

395 return float(obj)

396 elif isinstance(obj, l10n.Locale): 396 ↛ 398line 396 didn't jump to line 398 because the condition on line 396 was always true

397 return l10n.locale2str(obj)

398 elif isinstance(obj, (pathlib.PurePath, Path)):

399 return str(obj)

400 elif isinstance(obj, uuid.UUID):

401 return str(obj)

402 else:

403 msg = f'{obj} type {type(obj)} not serializable'

404 LOGGER.error(msg)

405 raise TypeError(msg)

406

407

408def is_url(urlstring: str) -> bool:

409 """

410 Validation function that determines whether a candidate URL should be

411 considered a URI. No remote resource is obtained; this does not check

412 the existence of any remote resource.

413

414 :param urlstring: `str` to be evaluated as candidate URL.

415

416 :returns: `bool` of whether the URL looks like a URL.

417 """

418 try:

419 result = urlparse(urlstring)

420 return bool(result.scheme and result.netloc)

421 except ValueError:

422 return False

423

424

425def render_j2_template(config: dict, tpl_config: dict, template: Path,

426 data: dict, locale_: str = None) -> str:

427 """

428 render Jinja2 template

429

430 :param config: dict of configuration

431 :param tpl_config: dict of template configuration

432 :param template: template (relative path)

433 :param data: dict of data

434 :param locale_: the requested output Locale

435

436 :returns: string of rendered template

437 """

438

439 template_paths = [TEMPLATESDIR, '.']

440

441 locale_dir = config['server'].get('locale_dir', 'locale')

442 LOGGER.debug(f'Locale directory: {locale_dir}')

443

444 try:

445 templates = tpl_config['path']

446 template_paths.insert(0, templates)

447 LOGGER.debug(f'using custom templates: {templates}')

448 except (KeyError, TypeError):

449 LOGGER.debug(f'using default templates: {TEMPLATESDIR}')

450

451 env = Environment(loader=FileSystemLoader(template_paths),

452 extensions=['jinja2.ext.i18n'],

453 autoescape=select_autoescape())

454

455 env.filters['to_json'] = to_json

456 env.filters['format_datetime'] = format_datetime

457 env.filters['format_duration'] = format_duration

458 env.filters['human_size'] = human_size

459 env.globals.update(to_json=to_json)

460

461 env.filters['get_path_basename'] = get_path_basename

462 env.globals.update(get_path_basename=get_path_basename)

463

464 env.filters['get_breadcrumbs'] = get_breadcrumbs

465 env.globals.update(get_breadcrumbs=get_breadcrumbs)

466

467 env.filters['filter_dict_by_key_value'] = filter_dict_by_key_value

468 env.globals.update(filter_dict_by_key_value=filter_dict_by_key_value)

469

470 translations = Translations.load(locale_dir, [locale_])

471 env.install_gettext_translations(translations)

472

473 try:

474 template = env.get_template(template)

475 except TemplateNotFound:

476 LOGGER.debug(f'template {template} not found')

477 template_paths.remove(templates)

478 template = env.get_template(template)

479

480 return template.render(config=l10n.translate_struct(config, locale_, True),

481 data=data, locale=locale_, version=__version__)

482

483

484def get_mimetype(filename: str) -> str:

485 """

486 helper function to return MIME type of a given file

487

488 :param filename: filename (with extension)

489

490 :returns: MIME type of given filename

491 """

492

493 return mimetypes.guess_type(filename)[0]

494

495

496def get_breadcrumbs(urlpath: str) -> list:

497 """

498 helper function to make breadcrumbs from a URL path

499

500 :param urlpath: URL path

501

502 :returns: `list` of `dict` objects of labels and links

503 """

504

505 links = []

506

507 tokens = urlpath.split('/')

508

509 s = ''

510 for t in tokens:

511 if s:

512 s += '/' + t

513 else:

514 s = t

515 links.append({

516 'href': s,

517 'title': t,

518 })

519

520 return links

521

522

523def filter_dict_by_key_value(dict_: dict, key: str, value: str) -> dict:

524 """

525 helper function to filter a dict by a dict key

526

527 :param dict_: ``dict``

528 :param key: dict key

529 :param value: dict key value

530

531 :returns: filtered ``dict``

532 """

533

534 return {k: v for (k, v) in dict_.items() if v[key] == value}

535

536

537class ProcessExecutionMode(Enum):

538 sync_execute = 'sync-execute'

539 async_execute = 'async-execute'

540

541

542class RequestedProcessExecutionMode(Enum):

543 wait = 'wait'

544 respond_async = 'respond-async'

545

546

547class RequestedResponse(Enum):

548 raw = 'raw'

549 document = 'document'

550

551

552class JobStatus(Enum):

553 """

554 Enum for the job status options specified in the WPS 2.0 specification

555 """

556

557 # From the specification

558 accepted = 'accepted'

559 running = 'running'

560 successful = 'successful'

561 failed = 'failed'

562 dismissed = 'dismissed'

563

564

565@dataclass(frozen=True)

566class Subscriber:

567 """

568 Store subscriber URLs as defined in:

569

570 https://schemas.opengis.net/ogcapi/processes/part1/1.0/openapi/schemas/subscriber.yaml # noqa

571 """

572

573 success_uri: str

574 in_progress_uri: Optional[str]

575 failed_uri: Optional[str]

576

577

578def read_data(path: Union[Path, str]) -> Union[bytes, str]:

579 """

580 helper function to read data (file or network)

581 """

582

583 LOGGER.debug(f'Attempting to read {path}')

584

585 if isinstance(path, Path) or not path.startswith(('http', 's3')):

586 LOGGER.debug('local file on disk')

587 with Path(path).open('rb') as fh:

588 return fh.read()

589 else:

590 LOGGER.debug('network file')

591 with urlopen(path) as r:

592 return r.read()

593

594

595def url_join(*parts: str) -> str:

596 """

597 helper function to join a URL from a number of parts/fragments.

598 Implemented because urllib.parse.urljoin strips subpaths from

599 host urls if they are specified

600

601 Per https://github.com/geopython/pygeoapi/issues/695

602

603 :param parts: list of parts to join

604

605 :returns: str of resulting URL

606 """

607

608 return '/'.join([p.strip().strip('/') for p in parts]).rstrip('/')

609

610

611def get_envelope(coords_list: List[List[float]]) -> list:

612 """

613 helper function to get the envelope for a given coordinates

614 list through the Shapely API.

615

616 :param coords_list: list of coordinates

617

618 :returns: list of the envelope's coordinates

619 """

620

621 coords = [tuple(item) for item in coords_list]

622 polygon = Polygon(coords)

623 bounds = polygon.bounds

624 return [[bounds[0], bounds[3]],

625 [bounds[2], bounds[1]]]

626

627

628class UrlPrefetcher:

629 """ Prefetcher to get HTTP headers for specific URLs.

630 Allows a maximum of 1 redirect by default.

631 """

632

633 def __init__(self):

634 self._session = Session()

635 self._session.max_redirects = 1

636

637 def get_headers(self, url: str, **kwargs) -> CaseInsensitiveDict:

638 """ Issues an HTTP HEAD request to the given URL.

639 Returns a case-insensitive dictionary of all headers.

640 If the request times out (defaults to 1 second unless `timeout`

641 keyword argument is set), or the response has a bad status code,

642 an empty dictionary is returned.

643 """

644 kwargs.setdefault('timeout', 1)

645 kwargs.setdefault('allow_redirects', True)

646 try:

647 response = self._session.head(url, **kwargs)

648 response.raise_for_status()

649 except Exception: # noqa

650 return CaseInsensitiveDict()

651 return response.headers

652

653

654def bbox2geojsongeometry(bbox: list) -> dict:

655 """

656 Converts bbox values into GeoJSON geometry

657

658 :param bbox: `list` of minx, miny, maxx, maxy

659

660 :returns: `dict` of GeoJSON geometry

661 """

662

663 b = box(*bbox, ccw=False)

664 return geom_to_geojson(b)

665

666

667def get_from_headers(headers: dict, header_name: str) -> str:

668 """

669 Gets case insensitive value from dictionary.

670 This is particularly useful when trying to get

671 headers from Starlette and Flask without issue

672

673 :param headers: `dict` of request headers.

674 :param header_name: Name of request header.

675

676 :returns: `str` value of header

677 """

678

679 cleaned_headers = {k.strip().lower(): v for k, v in headers.items()}

680 return cleaned_headers.get(header_name.lower(), '')

681

682

683def get_choice_from_headers(headers: dict,

684 header_name: str,

685 all: bool = False) -> Union[str, List[str]]:

686 """

687 Gets choices from a request dictionary,

688 considering numerical ordering of preferences.

689 Supported are complex preference strings (e.g. "fr-CH, fr;q=0.9, en;q=0.8")

690

691 :param headers: `dict` of request headers.

692 :param header_name: Name of request header.

693 :param all: bool to return one or all header values.

694

695 :returns: Sorted choice or choices from header

696 """

697

698 # Select header of interest

699 header = get_from_headers(headers=headers, header_name=header_name)

700 if header == '':

701 return

702

703 # Parse choices, extracting optional q values (defaults to 1.0)

704 choices = []

705 for i, part in enumerate(header.split(',')):

706 match = re.match(r'^([^;]+)(?:;q=([\d.]+))?$', part.strip())

707 if match:

708 value, q_value = match.groups()

709 q_value = float(q_value) if q_value else 1.0

710

711 # Sort choices by q value and index

712 if 0 <= q_value <= 1: 712 ↛ 705line 712 didn't jump to line 705 because the condition on line 712 was always true

713 heappush(choices, (1 / q_value, i, value))

714

715 # Drop q value

716 sorted_choices = [choice[-1] for choice in choices]

717

718 # Return one or all choices

719 return sorted_choices if all else sorted_choices[0]

720

721

722def get_dataset_formatters(dataset: dict) -> dict:

723 """

724 Helper function to derive all formatters for a collection

725

726 :param dataset: `dict` of dataset resource definition

727

728 :returns: `dict` of formatters

729 """

730

731 dataset_formatters = {}

732 provider_type = get_provider_default(dataset['providers'])['type']

733

734 for key, value in PLUGINS['formatter'].items():

735 # workaround to keep items-based collections supporting CSV

736 if provider_type not in ['feature', 'record']: 736 ↛ 737line 736 didn't jump to line 737 because the condition on line 736 was never true

737 continue

738

739 df2 = load_plugin('formatter', {'name': key})

740 dataset_formatters[key] = df2

741

742 for df in dataset.get('formatters', []): 742 ↛ 743line 742 didn't jump to line 743 because the loop on line 742 never started

743 df2 = load_plugin('formatter', df)

744 dataset_formatters[df2.name] = df2

745

746 return dataset_formatters

747

748

749def remove_url_auth(url: str) -> str:

750 """

751 Provide a RFC1738 URL without embedded authentication

752 :param url: RFC1738 URL

753 :returns: RFC1738 URL without authentication

754 """

755

756 u = urlparse(url)

757 auth = f'{u.username}:{u.password}@'

758 return url.replace(auth, '')

759

760

761def is_request_allowed(url: str, allow_internal: bool = False) -> bool:

762 """

763 Test whether an HTTP request is allowed to be executed

764

765 :param url: `str` of URL

766 :param allow_internal: `bool` of whether internal requests are

767 allowed (default `False`)

768

769 :returns: `bool` of whether HTTP request execution is allowed

770 """

771

772 is_allowed = False

773

774 u = urlparse(url)

775

776 ip = socket.gethostbyname(u.hostname)

777

778 is_private = ipaddress.ip_address(ip).is_private

779

780 if not is_private:

781 is_allowed = True

782 if is_private and allow_internal: 782 ↛ 783line 782 didn't jump to line 783 because the condition on line 782 was never true

783 is_allowed = True

784

785 return is_allowed