Coverage for src/lektor_ng/metaformat.py: 87%

71 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-09-05 15:10 +0000

1def _line_is_dashes(line): 

2 line = line.strip() 

3 return line == "-" * len(line) and len(line) >= 3 

4 

5 

6def _process_buf(buf): 

7 for idx, line in enumerate(buf): 

8 if _line_is_dashes(line): 

9 line = line[1:] 

10 buf[idx] = line 

11 

12 if buf and buf[-1][-1:] == "\n": 

13 buf[-1] = buf[-1][:-1] 

14 

15 return buf[:] 

16 

17 

18def tokenize(iterable, interesting_keys=None, encoding=None): 

19 """This tokenizes an iterable of newlines as bytes into key value 

20 pairs out of the lektor bulk format. By default it will process all 

21 fields, but optionally it can skip values of uninteresting keys and 

22 will instead yield `None`. The values are left as list of decoded 

23 lines with their endings preserved. 

24 

25 This will not perform any other processing on the data other than 

26 decoding and basic tokenizing. 

27 """ 

28 key = [] 

29 buf = [] 

30 want_newline = False 

31 is_interesting = True 

32 

33 def _flush_item(): 

34 the_key = key[0] 

35 if not is_interesting: 

36 value = None 

37 else: 

38 value = _process_buf(buf) 

39 del key[:], buf[:] 

40 return the_key, value 

41 

42 if encoding is not None: 

43 iterable = (x.decode(encoding, "replace") for x in iterable) 

44 

45 for line in iterable: 

46 line = line.rstrip("\r\n") + "\n" 

47 

48 if line.rstrip() == "---": 

49 want_newline = False 

50 if key: 

51 yield _flush_item() 

52 elif key: 

53 if want_newline: 

54 want_newline = False 

55 if not line.strip(): 

56 continue 

57 if is_interesting: 

58 buf.append(line) 

59 else: 

60 bits = line.split(":", 1) 

61 if len(bits) == 2: 

62 key = [bits[0].strip()] 

63 if interesting_keys is None: 

64 is_interesting = True 

65 else: 

66 is_interesting = key[0] in interesting_keys 

67 if is_interesting: 

68 first_bit = bits[1].strip("\t ") 

69 if first_bit.strip(): 

70 buf = [first_bit] 

71 else: 

72 buf = [] 

73 want_newline = True 

74 

75 if key: 

76 yield _flush_item() 

77 

78 

79def serialize(iterable, encoding=None): 

80 """Serializes an iterable of key value pairs into a stream of 

81 string chunks. If an encoding is provided, it will be encoded into that. 

82 

83 This is primarily used by the editor to write back data to a source file. 

84 """ 

85 

86 def _produce(item, escape=False): 

87 if escape: 

88 if _line_is_dashes(item): 

89 item = "-" + item 

90 if encoding is not None: 

91 item = item.encode(encoding) 

92 return item 

93 

94 for idx, (key, value) in enumerate(iterable): 

95 value = value.replace("\r\n", "\n").replace("\r", "\n") 

96 if idx > 0: 

97 yield _produce("---\n") 

98 if "\n" in value or value.strip("\t ") != value: 

99 yield _produce(key + ":\n") 

100 yield _produce("\n") 

101 for line in value.splitlines(True): 

102 yield _produce(line, escape=True) 

103 yield _produce("\n") 

104 else: 

105 yield _produce(f"{key}: {value}\n")