# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from copy import copy
import os
import textwrap
import typing as t
from cylc.flow.parsec.util import itemstr
TRACEBACK_WRAPPER = textwrap.TextWrapper()
[docs]
class ParsecError(Exception):
"""Generic exception for Parsec errors."""
schd_expected: bool = False
"""Set this flag to True on the exception if it is anticipated during
Cylc Scheduler run (apart from loading of config we do not expect
ParsecErrors during runtime)."""
[docs]
class ItemNotFoundError(ParsecError, KeyError):
"""Error raised for missing configuration items."""
def __init__(self, item):
self.item = item
def __str__(self):
return f'You have not set \"{self.item}\" in this config.'
[docs]
class InvalidConfigError(ParsecError, KeyError):
"""Error raised for missing configuration items."""
def __init__(self, item, specname):
self.item = item
self.specname = specname
def __str__(self):
return (
f'"{self.item}" is not a valid '
f'configuration for {self.specname}.'
)
[docs]
class NotSingleItemError(ParsecError, TypeError):
"""Error raised if an iterable is given where an item is expected."""
def __init__(self, item):
self.item = item
def __str__(self):
return f'not a singular item: {self.item}'
[docs]
class FileParseError(ParsecError):
"""Error raised when attempting to read in the config file(s).
Args:
reason:
Description of error.
err_type:
Classification of error (e.g. Jinja2Error).
help_lines:
Additional info to include in the exception.
lines:
(preferred) Dictionary in the format
{filename: [context_line, ..., error_line]}
index:
The line number of the error in the config (counting from the
shebang line *not* the first line).
line:
The line of the error in the config.
fpath:
The path to the file containing the error.
"""
def __init__(
self,
reason: str,
index: t.Optional[int] = None,
line: t.Optional[str] = None,
lines: t.Optional[t.Dict[str, t.List[str]]] = None,
err_type: t.Optional[str] = None,
fpath: t.Optional[str] = None,
help_lines: t.Optional[t.Iterable[str]] = None,
):
self.reason = reason
self.line_num = index + 1 if index is not None else None
self.line = line
self.lines = lines
self.err_type = err_type
self.fpath = fpath
self.help_lines = help_lines or []
def __str__(self) -> str:
msg = ''
msg += self.reason
if self.line_num is not None or self.fpath:
temp = []
if self.fpath:
temp.append(f'in {self.fpath}')
if self.line_num is not None:
temp.append(f'line {self.line_num}')
msg += f' ({" ".join(temp)})'
if self.line:
msg += ":\n " + self.line.strip()
if self.lines:
for filename, lines in self.lines.items():
msg += f'\nFile {filename}\n ' + '\n '.join(lines)
msg += "\t<--"
if self.err_type:
msg += ' %s' % self.err_type
help_lines = list(self.help_lines)
if self.line_num:
# TODO - make 'view' function independent of cylc:
help_lines.append("line numbers match 'cylc view -p'")
for help_line in help_lines:
msg += f'\n({help_line})'
return msg
[docs]
class TemplateVarLanguageClash(FileParseError):
"""Multiple workflow configuration templating engines configured."""
[docs]
class EmPyError(FileParseError):
"""Wrapper class for EmPy exceptions."""
[docs]
class Jinja2Error(FileParseError):
"""Wrapper class for Jinja2 exceptions.
Args:
exception:
The exception being re-raised
lines:
Dictionary in the format
{filename: [context_line, ..., error_line]}
filename:
Alternative to "lines" where less detail is available.
"""
def __init__(
self,
exception: Exception,
lines: t.Optional[t.Dict[str, t.List[str]]] = None,
filename: t.Optional[str] = None,
):
# extract the first sentence of exception
msg: str = str(exception)
try:
msg, tail = msg.split('. ', 1)
except ValueError:
tail = ''
else:
msg += '.'
tail = tail.strip()
# append the filename e.g. for a Jinja2 template
if filename:
msg += f'\nError in file "{filename}"'
# append the rest of the exception
if tail:
msg += '\n' + '\n'.join(TRACEBACK_WRAPPER.wrap(tail))
FileParseError.__init__(
self,
msg,
lines=lines,
err_type=exception.__class__.__name__
)
[docs]
class IncludeFileNotFoundError(ParsecError):
"""Error raised for missing include files."""
def __init__(self, flist):
"""Missing include file error.
E.g. for [DIR/top.cylc, DIR/inc/sub.cylc, DIR/inc/gone.cylc]
"Include-file not found: inc/gone.cylc via inc/sub.cylc from
DIR/top.cylc"
"""
rflist = copy(flist)
top_file = rflist[0]
top_dir = os.path.dirname(top_file) + '/'
rflist.reverse()
msg = rflist[0].replace(top_dir, '')
for f in rflist[1:-1]:
msg += ' via %s' % f.replace(top_dir, '')
msg += ' from %s' % top_file
ParsecError.__init__(self, msg)
[docs]
class UpgradeError(ParsecError):
"""Error raised upon fault in an upgrade operation."""
[docs]
class ValidationError(ParsecError):
"""Generic exception for invalid configurations."""
def __init__(self, keys, value=None, msg=None, exc=None, vtype=None,
key=None):
self.keys = keys
self.value = value
self.msg = msg
self.exc = exc
self.vtype = vtype
self.key = key
def __str__(self):
msg = ''
if self.vtype:
msg += f'(type={self.vtype}) '
if self.key:
msg += itemstr(self.keys, self.key)
elif self.value:
msg += itemstr(self.keys[:-1], self.keys[-1], value=self.value)
if self.msg or self.exc:
msg += (
f' - ({self.exc or ""}'
f'{": " if (self.exc and self.msg) else ""}'
f'{self.msg or ""})'
)
return msg
[docs]
class IllegalValueError(ValidationError):
"""Bad setting value."""
def __init__(self, vtype, keys, value, exc=None, msg=None):
ValidationError.__init__(
self, keys, vtype=vtype, value=value, exc=exc, msg=msg)
[docs]
class ListValueError(IllegalValueError):
"""Bad setting value, for a comma separated list."""
def __init__(self, keys, value, msg=None, exc=None):
IllegalValueError.__init__(
self, 'list', keys, value, exc=exc, msg=msg)
[docs]
class IllegalItemError(ValidationError):
"""Bad setting section or option name."""
def __init__(self, keys, key, msg=None, exc=None):
ValidationError.__init__(self, keys, key=key, exc=exc, msg=msg)