Fixed license error

This commit is contained in:
2026-02-23 22:37:13 +01:00
parent 8a21c2e4f0
commit 2f467a4128
6 changed files with 60 additions and 46 deletions
+30 -30
View File
@@ -78,7 +78,7 @@ log_end_offsets = {}
def format_topic_partitions(topic_name: str, partitions: set) -> list:
'''
Formats the topic partitions.
:param topic_name: The name of the topic, e. g. ``test-topic``
:type topic_name: str
:param partitions: The partitions, e. g. ``{1, 2, 4}``
@@ -92,7 +92,7 @@ def format_topic_partitions(topic_name: str, partitions: set) -> list:
def mktopic(topic_name: str):
'''
Creates a topic.
:param topic_name: The name of the topic
:type topic_name: str
'''
@@ -107,7 +107,7 @@ def mktopic(topic_name: str):
def rmtopic(topic_name: str):
'''
Removes a topic.
:param topic_name: The topic name
:type topic_name: str
@@ -130,7 +130,7 @@ def rmtopic(topic_name: str):
def mktopicpart(topic_name: str, partition_number: int):
'''
Creates a partition of a topic.
:param topic_name: The topic name
:type topic_name: str
:param partition_number: The partition number
@@ -153,7 +153,7 @@ def mktopicpart(topic_name: str, partition_number: int):
def rmtopicpart(topic_name: str, partition_number: int):
'''
Removes a partition of a topic.
:param topic_name: The topic name
:type topic_name: str
:param partition_number: The partition number
@@ -194,7 +194,7 @@ def _create_base_segment_files(topics: dict):
def get_segment_base_timestamp(topic: str, segment: int) -> int:
'''
Returns the base timestamp of a segment.
:param topic: The topic
:type topic: str
:param segment: The segment base offset
@@ -216,7 +216,7 @@ def get_segment_base_timestamp(topic: str, segment: int) -> int:
timestamp = None
else:
timestamp = int.from_bytes(timestamp_bytes)
return timestamp
def get_log_end_offsets(topics: dict):
@@ -247,7 +247,7 @@ def get_log_end_offsets(topics: dict):
def init(config: dict = {'segment_max_size': 1024 * 256, 'data_dir': './data'}):
'''
Initializes the module. Calling this function is required before using most of the functions of the module.
:param config: The configuration
:type config: dict
'''
@@ -269,7 +269,7 @@ def get_next_lower_index_entry(offset: int):
def validate_topic_partition(partition_number: str | int) -> bool:
'''
Validates a topic partition number format.
:param partition_number: The partition number
:type partition_number: str | int
@@ -283,16 +283,16 @@ def validate_topic_partition(partition_number: str | int) -> bool:
elif not isinstance(partition_number, int):
return False
if not (0 < partition_number < 100):
return False
return True
def validate_topic_name(topic_name: str) -> bool:
'''
Validates a topic name.
:param topic_name: The topic name
:type topic_name: str
@@ -301,16 +301,16 @@ def validate_topic_name(topic_name: str) -> bool:
'''
if not isinstance(topic_name, str):
return False
if not _TOPIC_NAME_REGEX.match(topic_name):
return False
if len(topic_name) == 0 or len(topic_name) > 255:
return False
if topic_name.startswith('.') or topic_name.endswith('.') or topic_name.startswith('..') or topic_name.endswith('..'):
return False
return True
def validate_topic_format(topic: str):
@@ -319,11 +319,11 @@ def validate_topic_format(topic: str):
_, formatted_partition_number = topic.split(TOPIC_PARTITION_SEPARATOR)
if len(formatted_partition_number) != 2 or not formatted_partition_number.isdigit():
raise ValueError('invalid partition number format')
def check_topic_exists(topic: str):
'''
Checks whether a topic exists.
:param topic: The topic name
:type topic: str
@@ -368,7 +368,7 @@ def _unpack_record_headers(buffer: io.BytesIO, number: int = 256):
key = buffer.read(key_length)
if len(key) < key_length:
break
value_length_bytes = buffer.read(4)
if len(value_length_bytes) < 4:
break
@@ -384,7 +384,7 @@ def _unpack_record_headers(buffer: io.BytesIO, number: int = 256):
def unpack_record_headers(packed: bytes) -> dict:
'''
Unpacks the record headers from bytes.
:param packed: The packed headers
:type packed: bytes
@@ -397,7 +397,7 @@ def unpack_record_headers(packed: bytes) -> dict:
def create_record(topic: str, timestamp: int, record_data: bytes, key: bytes = b'', compression_type: int = 0, headers: dict = {}):
'''
Creates a record.
:param topic: The topic name
:type topic: str
:param timestamp: The timestamp.
@@ -416,7 +416,7 @@ def create_record(topic: str, timestamp: int, record_data: bytes, key: bytes = b
topic_name, formatted_partition_number = topic.split(TOPIC_PARTITION_SEPARATOR)
check_topic_exists(topic)
topic_dir = f'{DATA_DIR}/topics/{topic_name}{TOPIC_PARTITION_SEPARATOR}{formatted_partition_number}/'
utils.create_file_if_not_exists(topic_dir + '0' * 16 + '.log')
segments = sorted([int(f.split('.')[0], 16) for f in os.listdir(topic_dir) if f.endswith('.log') and utils.is_number(f.split('.')[0], 16) and len(f.split('.')[0]) == 16])
@@ -447,7 +447,7 @@ def create_record(topic: str, timestamp: int, record_data: bytes, key: bytes = b
record += key
record += len(value).to_bytes(4)
record += value
record = len(record).to_bytes(8) + record # Prepend Record Size
@@ -460,7 +460,7 @@ def create_record(topic: str, timestamp: int, record_data: bytes, key: bytes = b
utils.create_file_if_not_exists(topic_dir + f'{last_segment:016x}.index')
utils.create_file_if_not_exists(topic_dir + f'{last_segment:016x}.timeindex')
if log_end_offsets[topic] % 1024 == 0:
index = log_end_offsets[topic].to_bytes(8) + os.path.getsize(topic_dir + f'{last_segment:016x}.log').to_bytes(4)
timeindex = timestamp.to_bytes(8) + log_end_offsets[topic].to_bytes(8)
@@ -476,7 +476,7 @@ def create_record(topic: str, timestamp: int, record_data: bytes, key: bytes = b
async def fetch_records(topic: str, start_type: int, start: int, max_bytes: int, send_block_function: FunctionType | MethodType):
'''
Fetches records from the topic and sends it using ``send_block_function``
:param topic: The topic name
:type topic: str
:param start_type: The start type
@@ -509,7 +509,7 @@ async def fetch_records(topic: str, start_type: int, start: int, max_bytes: int,
if len(timestamp_bytes) < 8:
break
timestamp = int.from_bytes(timestamp_bytes)
offset_bytes = f.read(8)
if len(offset_bytes) < 8:
break
@@ -603,7 +603,7 @@ async def fetch_records(topic: str, start_type: int, start: int, max_bytes: int,
await send_block_function(b'\x00') # EOF
class Topic:
@@ -612,11 +612,11 @@ class Topic:
def __repr__(self):
return f'Topic({self.topic_name})'
@property
def segments(self):
return [Segment(self.topic_name, bo) for bo in sorted([int(s.split('.')[0], 16) for s in os.listdir(f'{DATA_DIR}/topics/{self.topic_name}') if s.endswith('.log') and utils.is_number(s.split('.')[0], 16) and len(s.split('.')[0]) == 16])]
class Segment:
def __init__(self, topic: str, base_offset: int):
self.topic = topic
@@ -633,7 +633,7 @@ class Segment:
timestamp = None
else:
timestamp = int.from_bytes(timestamp_bytes)
return timestamp
class FileCorruptWarning(UserWarning): ...