If you want to search for a specific substring (pattern) within a string, you can use the str.find() or str.index() methods. If the pattern is found, both return the index of its first occurrence. If there is no match, str.find() returns -1, while str.index() raises a ValueError exception.
However, these two methods can only be called on strings of type str. Therefore, they cannot be used when:
- you need to search for a specific pattern in a large text file whose contents cannot be loaded into memory all at once, or
- you need to perform the search on a character sequence whose characters are supplied one at a time by an iterable object.
Since the standard library does not provide a function or method specifically designed for searching in these situations, we need to implement our own search function. We will create one function that accepts an iterable and another that accepts a file object.
These search functions should meet the following requirements:
- memory usage can be controlled during the search,
- the positions of all occurrences are provided,
- overlapping matches are detected.
Because of the last two requirements, it makes sense to implement the functions as generator functions. For a long sequence with many overlapping matches, the resulting sequence of indices can itself be very long. A generator function, however, gives us a generator object from which we can retrieve the positions one at a time. If we only need the first occurrence, we simply retrieve only one value from the generator.
As our functions search for all occurrences, it is useful to find the positions of overlapping matches, because there may be use cases where these are needed. If only non-overlapping matches are required, they can be filtered out relatively easily from the resulting sequence of indices.
Memory usage can be controlled by processing the incoming characters in blocks (chunks) of a configurable size.
This means that the incoming characters are grouped into strings of the specified chunk size (or, when searching a file, the file is read in chunks of this size), and the actual search is performed within each chunk. We’ll do this by calling find() on each chunk. The str.find() method is optimized and implemented in C, so it performs the search very quickly—much faster than an equivalent implementation written in Python. However, we have to keep in mind that str.find() only finds the first occurrence.
When processing the input stream chunk by chunk, we must account for the possibility that a pattern may span the boundary between two consecutive blocks. Therefore, the search cannot actually be performed on a single block. Instead, we need to search a string (the search string) formed by combining the appropriate trailing portion of the previous chunk with the new one.
When determining how much of the end of the previous chunk to include in the search string, we have to consider the longest possible overlap. This occurs when only a single character of the pattern is at the beginning of the next chunk, while all the remaining characters are at the end of the current chunk. Thus, the beginning of each search string overlaps with the last pattern length-1 characters of the previous chunk. This is illustrated in the following image.
The definitions of the find_in_iterable() and find_in_file() functions, implemented according to the principles described above, are shown below. The detailed comments in the code help explain how the functions work.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 |
from collections.abc import Iterable, Iterator from itertools import islice def find_in_iterable(s: Iterable[str], sub: str, start: int | None = None, end: int | None = None, chunk_size: int = 10000) -> Iterator[int]: """Return a generator object that yields the positions where the ‘sub’ substring is found in the character sequence provided by the iterable object 's'. Characters are not loaded all at once, but in chunks of length 'chunk_size'. 'chunk_size' must be a positive integer such that the length of 'sub' is at most one greater than it. The 'start' and 'end' parameters have the same meaning as their counterparts in str.find(): the search starts at the 'start' position but does not include the 'end' position. If 'start' is None, then the search starts at the beginning of the file. If 'end' is None, the search continues until the end of the file. The values of 'start' and 'end' cannot be negative. Parameters: s: An iterable object that returns characters. In each step of the iteration, it returns one character. sub: The substring to search for. start: The starting position of the search. If its value is the default None, the search starts at position 0. end: The position marking the end of the search range, beyond which no search is performed. If its value is the default None, search continues as long as the iterable object can yield characters. chunk_size: The length of the character segment retrieved from the iterable object at a time. Return value: An iterator that yields the positions where 'sub' occurs within the specified range, or -1 if the substring is not found. A ValueError exception is raised if: - the value of 'start' or 'end' is not 'None' and is not a non-negative integer, - 'start' is greater than 'end', - 'sub' is an empty string, - the value of 'chunk_size' is less than the allowed minimum. """ # An exception is thrown if the searched substring is an empty string. if sub == '': raise ValueError(f'"sub" cannot be an empty string.') # Check that the start and end positions that are not 'None' are non-negative integers. def check_index_nonnegative(position_index: int | None, parameter_name: str) -> None: if position_index is not None and (not type(position_index) is int or position_index < 0): raise ValueError(f'Az "{parameter_name}" argumentum nemnegatív egész szám kell, hogy legyen') check_index_nonnegative(start, 'start') check_index_nonnegative(end, 'end') min_chunk_size = max(1, len(sub) - 1) if type(chunk_size) is not int or chunk_size < min_chunk_size: raise ValueError(f'"chunk_size" must be a positive integer with a minimum value of {min_chunk_size}.') start = 0 if start is None else start # Check whether the start position is less than the end position. if end is not None and start > end: raise ValueError(f'"start" cannot be greater than "end"') # Create an iterator that returns characters between the start and end positions. s_iter: Iterator[str] = islice(s, start, end) # The maximum number of characters taken from the end of the previous chunk. overlap_size = len(sub) - 1 overlap = '' # The start position of the current chunk in 's'. chunk_start = start sub_found = False while True: # Create a string of length 'chunk_size'. chunk = ''.join(islice(s_iter, chunk_size)) if not chunk: break # Concatenate the characters kept from the end of the previous chunk and the current chunk. # This is necessary because the substring may be located at the boundary between two consecutive chunks. search_chunk = overlap + chunk # Search for occurrences of 'sub' in the current search section. index_in_search_chunk = 0 while True: index_in_search_chunk = search_chunk.find(sub, index_in_search_chunk, None) # If a match is found, the position within the search section is converted to the position # within the sequence 's'. if index_in_search_chunk != -1: absolute_position = chunk_start - len(overlap) + index_in_search_chunk # Only the match that begins within the current search range is valid. if absolute_position >= start: sub_found = True yield absolute_position index_in_search_chunk += 1 else: break # If there are no matches, keep the end of the current chunk for the next search, and # concatenate it to the beginning of the new chunk. We need at most len(sub) - 1 characters, because # if 'sub' extends beyond the chunk boundary, at least one of its characters is in the current chunk. overlap = chunk[-overlap_size:] if overlap_size else '' # Set the start position of the next chunk. chunk_start += len(chunk) # No matches found. if not sub_found: yield -1 |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
from collections.abc import Iterator def find_in_file(file_object, sub: str, start: int | None = None, end: int | None = None, chunk_size: int = 10000) -> Iterator[int]: """Return a generator object that yields the positions where the 'sub' substring is found in the 'file_object' file. The characters are not read from the file all at once, but in chunks of length 'chunk_size'. The value of 'chunk_size' must be a positive integer such that the length of 'sub' is at most one greater than it. The 'start' and 'end' parameters have the same meaning as their counterparts in str.find(): the search starts at the 'start' position but does not include the 'end' position. If 'start' is None, then the search starts at the beginning of the file. If 'end' is None, the search continues until the end of the file. The values of 'start' and 'end' cannot be negative. Parameters: file_object: The text file opened for reading in which the search is performed. sub: The substring to search for. start: The starting position of the search. If its value is the default None, the search starts at position 0. end: The position marking the end of the search range, beyond which no search is performed. If its value is the default None, the search continues to the end of the file. chunk_size: The length of the character segment read from the file at a time. Return value: An iterator that yields the positions where 'sub' occurs within the specified range, or -1 if the substring is not found. A ValueError exception is raised if: - 'file_object' is not a file object opened for reading. - the value of 'start' or 'end' is not 'None' and is not a non-negative integer, - 'start' is greater than 'end', - 'sub' is an empty string, - the value of 'chunk_size' is less than the allowed minimum. """ if not file_object.readable(): raise ValueError('The first argument must be a file object opened for reading.') # An exception is thrown if the searched substring is an empty string. if sub == '': raise ValueError(f'"sub" cannot be an empty string.') # Check that the start and end positions that are not 'None' are non-negative integers. def check_index_nonnegative(position_index: int | None, parameter_name: str) -> None: if position_index is not None and (not type(position_index) is int or position_index < 0): raise ValueError(f'The argument "{parameter_name}" must be a non-negative integer.') check_index_nonnegative(start, 'start') check_index_nonnegative(end, 'end') min_chunk_size = max(1, len(sub) - 1) if type(chunk_size) is not int or chunk_size < min_chunk_size: raise ValueError(f'"chunk_size" must be a positive integer with a minimum value of {min_chunk_size}.') start = 0 if start is None else start # Check whether the start position is less than the end position. if end is not None and start > end: raise ValueError(f'"start" cannot be greater than "end"') position = 0 # Holds the starting position of the next chunk in the file. # Reading is done in chunks until the start position is reached, and the data is discarded. # This is necessary because it eliminates the need to keep the entire portion of the file # before the start position in memory. while position < start: chunk = file_object.read(min(chunk_size, start - position)) if not chunk: # The end of the file was reached before the start position. yield -1 return position += len(chunk) overlap = "" sub_found = False while end is None or position < end: read_size = chunk_size if end is not None: read_size = min(read_size, end - position) if not (chunk := file_object.read(read_size)): break # Concatenate the characters kept from the end of the previous chunk and the current chunk. # This is necessary because the substring may be located at the boundary between two consecutive chunks. search_chunk = overlap + chunk # Search for occurrences of 'sub' in the current search section. index_in_search_chunk = 0 while True: index_in_search_chunk = search_chunk.find(sub, index_in_search_chunk, None) # If a match is found, the position within the search section is converted to the position # within the entire file. if index_in_search_chunk != -1: found_position = position - len(overlap) + index_in_search_chunk # The start position of the match is in the range [start, end), and the entire 'sub' # also fits within the search range. if end is None or found_position + len(sub) <= end: sub_found = True yield found_position index_in_search_chunk += 1 else: break # If there are no matches, keep the end of the current chunk for the next search, and # concatenate it to the beginning of the new chunk. We need at most len(sub) - 1 characters # because if 'sub' extends beyond the chunk boundary, at least one of its characters is in the # current chunk. overlap_size = len(sub) - 1 overlap = chunk[-overlap_size:] if overlap_size else '' # Set the start position of the next chunk. position += len(chunk) # No matches found. if not sub_found: yield -1 |
To test the functions, we need large files whose size can be specified in advance and in which we can place the string being searched for—the pattern—at a specified position. All other characters are generated randomly. To test the search with more than just ASCII characters, we also include accented letters (vowels) both in the pattern and among the randomly generated characters.
For each file size, we measure the search time under worst-case conditions. To achieve this, we place the pattern at the very end of the file. With sufficiently long sequences, however, the pattern may also occur by chance within the randomly generated section. It is therefore useful to estimate how many occurrences we can expect for a given sequence length and pattern length. This allows us to determine the character sequence lengths for which the pattern is almost certain to occur only at the end of the file, as well as the length beyond which multiple occurrences can be expected. In the latter case, our functions are expected to return the positions of all occurrences. This approach also gives us an opportunity to test our assumptions about the random distribution of characters.
The derivation of the formulas needed to design the large files containing long character sequences can be followed in the images below.
The definitions of the functions that generate the files and character stream using these formulas, along with the code that creates the files, are shown below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
import random import string from collections.abc import Iterator def gen_random_chars(num_chars: int, pattern='', index: int | None = None) -> Iterator[str]: """Yield 'num_chars' randomly generated characters one at a time. Starting at 'index'(inclusive), characters from 'pattern' are yielded. If 'index' is None, the 'pattern' characters are yielded at the end. Generated characters are drawn from Latin uppercase and lowercase letters, digits, and vowels with acute. """ vowels_with_acute = 'áéíóúüűöő' cap_vowels_with_acute = ''.join([c.upper() for c in vowels_with_acute]) base_chars = string.ascii_letters + string.digits + vowels_with_acute + cap_vowels_with_acute for i in range(num_chars - len(pattern)): if i == index: yield from pattern yield random.choice(base_chars) if index is None: yield from pattern def create_random_text_file(file_path: str, num_chars: int, pattern='', index: int | None = None) -> None: """Create a text file specified by ‘file_path’ that contains a total of num_chars random characters. The characters written to the file are generated by the gen_random_chars() function. Accordingly, the characters are selected from the set of lowercase and uppercase Latin letters, digits, and vowels with acute. Starting at position ‘index’ (inclusive) insert the characters from the ‘pattern’ argument. If ‘index’ is None, the ‘pattern’ is appended to the end of the file. """ char_gen = gen_random_chars(num_chars, pattern, index) with open(file_path, mode='w', encoding='utf-8', newline='') as f: for c in char_gen: f.write(c) def char_stream(file_path: str, chunk_size: int = 5000000) -> Iterator[str]: """Yield the contents of the specified text file character by character, without converting newline characters. To reduce the number of disk I/O operations and avoid having to load the entire file into memory, the file is read in chunks of length ‘chunk_size’. ‘chunk_size’ is the maximum number of characters to be read in a single read operation. """ with open(file_path, 'r', encoding='utf-8', newline='') as f: while chunk := f.read(chunk_size): yield from chunk def get_num_chars(file_path: str) -> int: """Return the number of characters stored in the specified text file.""" char_count = 0 with open(file_path, 'r', encoding='utf-8', newline='') as f: while chunk := f.read(1000000): char_count += len(chunk) return char_count def expected_bytes_for_chars(num_chars: int, pattern_length: int, num_2byte_chars_in_pattern: int) -> int: """Return the expected byte size of a string consisting of num_chars characters, assuming UTF-8 encoding. This string contains a substring of length pattern_length, consisting of characters represented by 1 or 2 bytes, such that the number of 2-byte characters is num_2byte_chars_in_pattern. The remaining characters are evenly distributed among the 62 1-byte ASCII characters (lowercase and uppercase letters and digits) and the 18 2-byte characters (the lowercase and uppercase vowels with acute). """ return round(1.225 * num_chars - 0.225 * pattern_length + num_2byte_chars_in_pattern) def expected_chars_for_bytes(num_bytes: int, pattern_length: int, num_2byte_chars_in_pattern: int) -> int: """Return the number of characters required to obtain num_bytes of UTF-8-encoded data. The character sequence contains a predefined string of pattern_length characters, each of which is represented by 2 bytes in UTF-8. The remaining characters are evenly distributed among the 62 1-byte ASCII characters (lowercase and uppercase letters and digits) and the 18 2-byte characters (the lowercase and uppercase vowels with acute). """ return round((num_bytes + 0.225 * pattern_length - num_2byte_chars_in_pattern) / 1.225) if __name__ == '__main__': # Create text files of various sizes in which the characters in the “pattern” are the last ones, and # the remaining characters are selected at random—with uniform distribution—from among lowercase and # uppercase Latin letters, digits, and vowels with acute. Mbyte = 1024 * 1024 pattern = 'KöprÜ' pattern_length = len(pattern) num_2byte_chars_in_pattern = sum(1 for c in pattern if len(c.encode()) == 2) factors = (1, 10, 100, 1000, 10000) file_sizes_in_bytes = [i * Mbyte for i in factors] file_chars_counts = [expected_chars_for_bytes(file_size, pattern_length, num_2byte_chars_in_pattern) for file_size in file_sizes_in_bytes] file_names = [f'random_text_{i}MB_pattern{pattern_length}.txt' for i in factors] for file_name, num_chars in zip(file_names, file_chars_counts): print(file_name, num_chars) create_random_text_file(file_name, num_chars, pattern, None) |
The code used to measure the search time of the find_in_iterable() and find_in_file() functions is the following.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 |
from time import perf_counter from source_generators import char_stream, get_num_chars def running_time_find_in_iterable(iterable, substring=''): start = perf_counter() indexes = list(find_in_iterable(iterable, substring)) stop = perf_counter() elapsed = stop - start print('Search time: {:.5f} s'.format(elapsed)) print('The string "{substring}" begins at positions {indexes}.') return elapsed, indexes, substring def running_time_find_in_file(filepath: str, substring: str): with open(filepath, 'r', encoding='UTF-8', newline='') as fileobj: start = perf_counter() indexes = list(find_in_file(fileobj, substring)) stop = perf_counter() elapsed = stop - start print('Search time: {:.5f} s'.format(elapsed)) print('The string "{substring}" begins at positions {indexes}.') return elapsed, indexes, substring # TEST factors = (1, 10, 100, 1000, 10000) file_paths = [f'random_text_{i}MB_pattern5.txt' for i in factors] substring = 'KöprÜ' for file_path in file_paths: num_chars = get_num_chars(file_path) print(f'\nThe number of characters stored in the file {file_path.name} is: {get_num_chars(file_path):_d}') print('---- find_in_iterable() ----') elapsed, indexes, substring = running_time_find_in_iterable(char_stream(file_path), substring) assert indexes[-1] + len(substring) == num_chars print('Number of occurrences of the searched substring:', len(indexes)) print('---- find_in_file() ----') elapsed, indexes, substring = running_time_find_in_file(file_path, substring) assert indexes[-1] + len(substring) == num_chars print('Number of occurrences of the searched substring:', len(indexes)) |
The results are summarized in a table below.
As we can see from the results, the search time increases linearly with the number of characters. This is what we would expect because for a given search pattern, the optimized search algorithm used by str.find() has a worst-case execution time that grows linearly with the length of the input sequence—even when the pattern occurs at the very end of the sequence.
For file-based searching, the linear relationship holds only in the 1–1000 MiB range. For a 10,000 MiB file, the search time is noticeably higher than we would expect based on the previous trend. The reason is that the smaller files fit entirely into the operating system’s page cache during testing, so the data was effectively read from fast RAM rather than from the storage device. The 10,000 MiB file, however, exceeded the available page-cache capacity. As a result, processing speed was no longer determined by the CPU and memory, but by the slower read speed of the physical storage device.
In the iterable-based version, the search time increased by an order of magnitude (approximately tenfold), but it still increased linearly even for the largest number of characters tested.
The slowdown is caused by processing each character at the Python level. When constructing the search strings, the combination of islice() and ''.join() requires the Python interpreter to perform a large number of internal operations, including function calls and temporary memory allocations. This also explains why the linear behavior is maintained across the entire range: the code becomes purely CPU-bound. Since a Python-level loop is much slower than C-based file reading, the processing speed here is no longer limited by the physical I/O performance of the storage device or by the size of the page cache. Instead, it is determined primarily by the CPU’s ability to perform computations at a steady rate.
Creating the functions presented in this article requires knowledge of a number of Python language features. These topics are covered in the e-book Python Knowledge Building Step by Step: From the Basics to Your First Desktop Application.