0
# High-Level File Interface
1
2
Python file-like interface with asynchronous methods, providing familiar patterns for developers while maintaining async/await compatibility. The high-level interface mimics Python's built-in file objects but with async methods.
3
4
## Capabilities
5
6
### async_open Function
7
8
High-level helper that mimics Python's built-in open() function but returns file-like objects with async methods.
9
10
```python { .api }
11
def async_open(
12
file_specifier: Union[str, Path, FileIOType],
13
mode: str = "r",
14
*args,
15
**kwargs
16
) -> Union[BinaryFileWrapper, TextFileWrapper]:
17
"""
18
Open file and return appropriate wrapper with async methods.
19
20
Args:
21
file_specifier: File path or existing file object
22
mode: File mode string (same as built-in open())
23
*args: Additional arguments passed to AIOFile constructor
24
**kwargs: Additional keyword arguments passed to AIOFile constructor
25
26
Returns:
27
BinaryFileWrapper for binary modes ('rb', 'wb', etc.)
28
TextFileWrapper for text modes ('r', 'w', etc.)
29
30
Raises:
31
ValueError: If args provided when wrapping existing file object
32
"""
33
```
34
35
### Base File Wrapper
36
37
Abstract base class providing common functionality for both binary and text file wrappers.
38
39
```python { .api }
40
class FileIOWrapperBase:
41
def __init__(self, afp: AIOFile, *, offset: int = 0):
42
"""
43
Initialize file wrapper.
44
45
Args:
46
afp: AIOFile instance to wrap
47
offset: Initial file position (0 for append mode sets to file end)
48
"""
49
50
def seek(self, offset: int) -> None:
51
"""
52
Set file position.
53
54
Args:
55
offset: New position in bytes
56
"""
57
58
def tell(self) -> int:
59
"""
60
Get current file position.
61
62
Returns:
63
Current position in bytes
64
"""
65
66
async def flush(self, sync_metadata: bool = False) -> None:
67
"""
68
Flush file buffers to disk.
69
70
Args:
71
sync_metadata: If True, sync metadata (slower but more complete)
72
"""
73
74
async def close(self) -> None:
75
"""Close the underlying file."""
76
77
def iter_chunked(self, chunk_size: int = 32768) -> Reader:
78
"""
79
Create chunked reader from current position.
80
81
Args:
82
chunk_size: Size of each chunk in bytes
83
84
Returns:
85
Reader instance for chunked iteration
86
"""
87
```
88
89
### Binary File Wrapper
90
91
File-like interface for binary files with async methods.
92
93
```python { .api }
94
class BinaryFileWrapper(FileIOWrapperBase):
95
def __init__(self, afp: AIOFile):
96
"""
97
Initialize binary file wrapper.
98
99
Args:
100
afp: AIOFile instance in binary mode
101
102
Raises:
103
ValueError: If AIOFile is not in binary mode
104
"""
105
106
async def read(self, length: int = -1) -> bytes:
107
"""
108
Read binary data from file.
109
110
Args:
111
length: Number of bytes to read (-1 for all remaining)
112
113
Returns:
114
Binary data as bytes
115
"""
116
117
async def write(self, data: bytes) -> int:
118
"""
119
Write binary data to file.
120
121
Args:
122
data: Binary data to write
123
124
Returns:
125
Number of bytes written
126
"""
127
128
async def readline(self, size: int = -1, newline: bytes = b"\n") -> bytes:
129
"""
130
Read one line from binary file.
131
132
Args:
133
size: Maximum bytes to read (-1 for no limit)
134
newline: Line separator bytes
135
136
Returns:
137
Line as bytes including separator
138
"""
139
```
140
141
### Text File Wrapper
142
143
File-like interface for text files with async methods and proper encoding handling.
144
145
```python { .api }
146
class TextFileWrapper(FileIOWrapperBase):
147
def __init__(self, afp: AIOFile):
148
"""
149
Initialize text file wrapper.
150
151
Args:
152
afp: AIOFile instance in text mode
153
154
Raises:
155
ValueError: If AIOFile is in binary mode
156
"""
157
158
@property
159
def encoding(self) -> str:
160
"""Text encoding of the file."""
161
162
async def read(self, length: int = -1) -> str:
163
"""
164
Read text data from file.
165
166
Args:
167
length: Number of characters to read (-1 for all remaining)
168
169
Returns:
170
Text data as string
171
"""
172
173
async def write(self, data: str) -> int:
174
"""
175
Write text data to file.
176
177
Args:
178
data: Text data to write
179
180
Returns:
181
Number of bytes written (encoded size)
182
"""
183
184
async def readline(self, size: int = -1, newline: str = "\n") -> str:
185
"""
186
Read one line from text file.
187
188
Args:
189
size: Maximum characters to read (-1 for no limit)
190
newline: Line separator string
191
192
Returns:
193
Line as string including separator
194
"""
195
```
196
197
## Usage Examples
198
199
### Basic Text File Operations
200
201
```python
202
import asyncio
203
from aiofile import async_open
204
205
async def text_operations():
206
# Read entire text file
207
async with async_open('document.txt', 'r') as file:
208
content = await file.read()
209
print(content)
210
211
# Write text file
212
async with async_open('output.txt', 'w') as file:
213
await file.write("Hello, async world!\n")
214
await file.write("Second line\n")
215
await file.flush() # Ensure data is written
216
217
asyncio.run(text_operations())
218
```
219
220
### Binary File Operations
221
222
```python
223
import asyncio
224
from aiofile import async_open
225
226
async def binary_operations():
227
# Read binary file
228
async with async_open('image.png', 'rb') as file:
229
header = await file.read(8) # Read PNG header
230
print(f"Header: {header.hex()}")
231
232
# Write binary file
233
async with async_open('output.bin', 'wb') as file:
234
data = bytes(range(256)) # Create sample binary data
235
await file.write(data)
236
237
asyncio.run(binary_operations())
238
```
239
240
### Line-by-Line Processing
241
242
```python
243
import asyncio
244
from aiofile import async_open
245
246
async def process_lines():
247
async with async_open('large_file.txt', 'r') as file:
248
# Method 1: Using async iteration (recommended)
249
async for line in file:
250
print(f"Line: {line.strip()}")
251
252
# Method 2: Manual readline
253
file.seek(0) # Reset to beginning
254
while True:
255
line = await file.readline()
256
if not line:
257
break
258
print(f"Manual: {line.strip()}")
259
260
asyncio.run(process_lines())
261
```
262
263
### File Position Management
264
265
```python
266
import asyncio
267
from aiofile import async_open
268
269
async def position_example():
270
async with async_open('data.txt', 'r+') as file:
271
# Read from beginning
272
data1 = await file.read(10)
273
print(f"Position after read: {file.tell()}")
274
275
# Seek to specific position
276
file.seek(5)
277
data2 = await file.read(5)
278
print(f"Data from position 5: {data2}")
279
280
# Write at current position
281
await file.write(" [INSERTED] ")
282
print(f"Position after write: {file.tell()}")
283
284
asyncio.run(position_example())
285
```
286
287
### Chunked Reading
288
289
```python
290
import asyncio
291
from aiofile import async_open
292
293
async def chunked_reading():
294
async with async_open('large_file.txt', 'r') as file:
295
# Create chunked reader
296
reader = file.iter_chunked(chunk_size=1024)
297
298
# Process chunks
299
async for chunk in reader:
300
print(f"Chunk size: {len(chunk)}")
301
# Process chunk...
302
303
asyncio.run(chunked_reading())
304
```
305
306
### Append Mode Operations
307
308
```python
309
import asyncio
310
from aiofile import async_open
311
312
async def append_operations():
313
# Append to existing file
314
async with async_open('log.txt', 'a') as file:
315
print(f"Starting position: {file.tell()}") # At end of file
316
await file.write("New log entry\n")
317
await file.flush()
318
319
asyncio.run(append_operations())
320
```
321
322
### Working with Existing File Objects
323
324
```python
325
import asyncio
326
from aiofile import async_open
327
328
async def wrap_existing_file():
329
# Wrap existing file object
330
with open('temp.txt', 'w+') as fp:
331
fp.write("Initial content")
332
fp.seek(0)
333
334
# Wrap with async interface
335
async_file = async_open(fp)
336
content = await async_file.read()
337
print(f"Content: {content}")
338
339
asyncio.run(wrap_existing_file())
340
```
341
342
### Error Handling
343
344
```python
345
import asyncio
346
from aiofile import async_open
347
348
async def error_handling():
349
try:
350
async with async_open('nonexistent.txt', 'r') as file:
351
content = await file.read()
352
except FileNotFoundError:
353
print("File not found")
354
355
try:
356
async with async_open('readonly.txt', 'w') as file:
357
await file.write("data")
358
except PermissionError:
359
print("Permission denied")
360
361
asyncio.run(error_handling())
362
```
363
364
## Context Manager Support
365
366
All file wrapper classes support async context managers for automatic resource management:
367
368
```python
369
# Automatic file opening and closing
370
async with async_open('file.txt', 'r') as file:
371
# File is automatically opened
372
data = await file.read()
373
# File is automatically closed and flushed when exiting context
374
```
375
376
The context manager ensures:
377
- File is opened when entering the context
378
- File is properly closed when exiting the context
379
- Data is flushed for writable files before closing
380
- Resources are cleaned up even if exceptions occur