Coverage for scheduler/base/scheduler_util.py: 96%

1""" 

2Implementation of essential functions and components for a `BaseJob`. 

3 

4Author: Jendrik A. Potyka, Fabian A. Preiss 

5""" 

6 

7import datetime as dt 

8from typing import Optional, Union, cast 

9 

10from scheduler.base.job import BaseJobType 

11from scheduler.base.timingtype import ( 

12 TimingCyclic, 

13 TimingDailyUnion, 

14 TimingJobUnion, 

15 TimingWeeklyUnion, 

16) 

17from scheduler.error import SchedulerError 

18 

19 

20def str_cutoff(string: str, max_length: int, cut_tail: bool = False) -> str: 

21 """ 

22 Abbreviate a string to a given length. 

23 

24 The resulting string will carry an indicator if it's abbreviated, 

25 like ``stri#``. 

26 

27 Parameters 

28 ---------- 

29 string : str 

30 String which is to be cut. 

31 max_length : int 

32 Max resulting string length. 

33 cut_tail : bool 

34 ``False`` for string abbreviation from the front, else ``True``. 

35 

36 Returns 

37 ------- 

38 str 

39 Resulting string 

40 """ 

41 if max_length < 1: 

42 raise ValueError("max_length < 1 not allowed") 

43 

44 if len(string) > max_length: 

45 pos = max_length - 1 

46 return string[:pos] + "#" if cut_tail else "#" + string[-pos:] 

47 

48 return string 

49 

50 

51def check_tzname(tzinfo: Optional[dt.tzinfo]) -> Optional[str]: 

52 """Composed of the datetime.datetime.tzname and the datetime._check_tzname methode.""" 

53 if tzinfo is None: 

54 return None 

55 name: Optional[str] = tzinfo.tzname(None) 

56 if not isinstance(name, str): 

57 raise SchedulerError(f"tzinfo.tzname() must return None or string, not {type(name)}") 

58 return name 

59 

60 

61def create_job_instance( 

62 job_class: type, 

63 timing: Union[TimingCyclic, TimingDailyUnion, TimingWeeklyUnion], 

64 **kwargs, 

65) -> BaseJobType: 

66 """Create a job instance from the given input parameters.""" 

67 if not isinstance(timing, list): 

68 timing_list = cast(TimingJobUnion, [timing]) 

69 else: 

70 timing_list = cast(TimingJobUnion, timing) 

71 

72 return cast( 

73 BaseJobType, 

74 job_class( 

75 timing=timing_list, 

76 **kwargs, 

77 ), 

78 )