0
# OS Path Operations
1
2
Async versions of os.path module functions for path manipulation and information gathering. These functions provide file system path introspection and validation without blocking the asyncio event loop.
3
4
## Capabilities
5
6
### Path Existence and Type Checking
7
8
Check whether paths exist and determine their types (file, directory, symbolic link, etc.).
9
10
```python { .api }
11
async def exists(path: str, *, loop = None, executor = None) -> bool:
12
"""
13
Check if path exists.
14
15
Parameters:
16
- path: Path to check
17
- loop: Event loop to use
18
- executor: Thread pool executor to use
19
20
Returns:
21
True if path exists, False otherwise
22
"""
23
24
async def isfile(path: str, *, loop = None, executor = None) -> bool:
25
"""
26
Check if path is a regular file.
27
28
Parameters:
29
- path: Path to check
30
- loop: Event loop to use
31
- executor: Thread pool executor to use
32
33
Returns:
34
True if path is a file, False otherwise
35
"""
36
37
async def isdir(path: str, *, loop = None, executor = None) -> bool:
38
"""
39
Check if path is a directory.
40
41
Parameters:
42
- path: Path to check
43
- loop: Event loop to use
44
- executor: Thread pool executor to use
45
46
Returns:
47
True if path is a directory, False otherwise
48
"""
49
50
async def islink(path: str, *, loop = None, executor = None) -> bool:
51
"""
52
Check if path is a symbolic link.
53
54
Parameters:
55
- path: Path to check
56
- loop: Event loop to use
57
- executor: Thread pool executor to use
58
59
Returns:
60
True if path is a symbolic link, False otherwise
61
"""
62
63
async def ismount(path: str, *, loop = None, executor = None) -> bool:
64
"""
65
Check if path is a mount point.
66
67
Parameters:
68
- path: Path to check
69
- loop: Event loop to use
70
- executor: Thread pool executor to use
71
72
Returns:
73
True if path is a mount point, False otherwise
74
"""
75
```
76
77
**Usage Example:**
78
79
```python
80
from aiofiles import ospath
81
82
# Check path existence and type
83
if await ospath.exists('example.txt'):
84
if await ospath.isfile('example.txt'):
85
print("example.txt is a file")
86
elif await ospath.isdir('example.txt'):
87
print("example.txt is a directory")
88
elif await ospath.islink('example.txt'):
89
print("example.txt is a symbolic link")
90
91
# Check for mount points
92
if await ospath.ismount('/mnt/external'):
93
print("/mnt/external is a mount point")
94
```
95
96
### File Size and Timestamps
97
98
Get file size and various timestamp information.
99
100
```python { .api }
101
async def getsize(path: str, *, loop = None, executor = None) -> int:
102
"""
103
Get file size in bytes.
104
105
Parameters:
106
- path: Path to file
107
- loop: Event loop to use
108
- executor: Thread pool executor to use
109
110
Returns:
111
File size in bytes
112
"""
113
114
async def getmtime(path: str, *, loop = None, executor = None) -> float:
115
"""
116
Get file modification time.
117
118
Parameters:
119
- path: Path to file
120
- loop: Event loop to use
121
- executor: Thread pool executor to use
122
123
Returns:
124
Modification time as seconds since epoch
125
"""
126
127
async def getatime(path: str, *, loop = None, executor = None) -> float:
128
"""
129
Get file access time.
130
131
Parameters:
132
- path: Path to file
133
- loop: Event loop to use
134
- executor: Thread pool executor to use
135
136
Returns:
137
Access time as seconds since epoch
138
"""
139
140
async def getctime(path: str, *, loop = None, executor = None) -> float:
141
"""
142
Get file creation time (Windows) or metadata change time (Unix).
143
144
Parameters:
145
- path: Path to file
146
- loop: Event loop to use
147
- executor: Thread pool executor to use
148
149
Returns:
150
Creation/change time as seconds since epoch
151
"""
152
```
153
154
**Usage Example:**
155
156
```python
157
import time
158
from aiofiles import ospath
159
160
# Get file information
161
size = await ospath.getsize('document.pdf')
162
print(f"File size: {size} bytes")
163
164
# Get timestamps
165
mtime = await ospath.getmtime('document.pdf')
166
atime = await ospath.getatime('document.pdf')
167
ctime = await ospath.getctime('document.pdf')
168
169
print(f"Modified: {time.ctime(mtime)}")
170
print(f"Accessed: {time.ctime(atime)}")
171
print(f"Created/Changed: {time.ctime(ctime)}")
172
173
# Check if file was modified recently
174
import asyncio
175
current_time = time.time()
176
if current_time - mtime < 3600: # Less than 1 hour ago
177
print("File was modified recently")
178
```
179
180
### File Comparison
181
182
Compare files to determine if they refer to the same file system object.
183
184
```python { .api }
185
async def samefile(path1: str, path2: str, *, loop = None, executor = None) -> bool:
186
"""
187
Check if two paths refer to the same file.
188
189
Parameters:
190
- path1: First path to compare
191
- path2: Second path to compare
192
- loop: Event loop to use
193
- executor: Thread pool executor to use
194
195
Returns:
196
True if paths refer to same file, False otherwise
197
"""
198
199
async def sameopenfile(fp1: int, fp2: int, *, loop = None, executor = None) -> bool:
200
"""
201
Check if two file descriptors refer to the same file.
202
203
Parameters:
204
- fp1: First file descriptor
205
- fp2: Second file descriptor
206
- loop: Event loop to use
207
- executor: Thread pool executor to use
208
209
Returns:
210
True if file descriptors refer to same file, False otherwise
211
"""
212
```
213
214
**Usage Example:**
215
216
```python
217
from aiofiles import ospath
218
219
# Check if paths refer to same file (handles symlinks, hardlinks)
220
same = await ospath.samefile('file.txt', 'link_to_file.txt')
221
if same:
222
print("Paths refer to the same file")
223
224
# Compare open file descriptors
225
with open('file1.txt') as f1, open('file2.txt') as f2:
226
same_fd = await ospath.sameopenfile(f1.fileno(), f2.fileno())
227
if same_fd:
228
print("File descriptors refer to same file")
229
```
230
231
### Path Manipulation
232
233
Transform and normalize file system paths.
234
235
```python { .api }
236
async def abspath(path: str, *, loop = None, executor = None) -> str:
237
"""
238
Get absolute path.
239
240
Parameters:
241
- path: Relative or absolute path
242
- loop: Event loop to use
243
- executor: Thread pool executor to use
244
245
Returns:
246
Absolute path string
247
"""
248
```
249
250
**Usage Example:**
251
252
```python
253
from aiofiles import ospath
254
255
# Convert relative path to absolute
256
relative_path = '../documents/file.txt'
257
absolute_path = await ospath.abspath(relative_path)
258
print(f"Absolute path: {absolute_path}")
259
260
# Get absolute path of current directory file
261
current_file_abs = await ospath.abspath('data.txt')
262
print(f"Current file absolute path: {current_file_abs}")
263
```
264
265
## Practical Usage Patterns
266
267
### File Validation Pipeline
268
269
```python
270
from aiofiles import ospath
271
import aiofiles.os
272
273
async def validate_file(path: str) -> dict:
274
"""Comprehensive file validation."""
275
result = {
276
'exists': await ospath.exists(path),
277
'is_file': False,
278
'is_dir': False,
279
'is_link': False,
280
'size': 0,
281
'modified': 0
282
}
283
284
if result['exists']:
285
result['is_file'] = await ospath.isfile(path)
286
result['is_dir'] = await ospath.isdir(path)
287
result['is_link'] = await ospath.islink(path)
288
289
if result['is_file']:
290
result['size'] = await ospath.getsize(path)
291
result['modified'] = await ospath.getmtime(path)
292
293
return result
294
295
# Usage
296
file_info = await validate_file('example.txt')
297
print(f"File info: {file_info}")
298
```
299
300
### Directory Processing
301
302
```python
303
from aiofiles import ospath
304
import aiofiles.os
305
306
async def process_directory(dir_path: str):
307
"""Process all files in directory."""
308
if not await ospath.exists(dir_path):
309
print(f"Directory {dir_path} does not exist")
310
return
311
312
if not await ospath.isdir(dir_path):
313
print(f"{dir_path} is not a directory")
314
return
315
316
files = await aiofiles.os.listdir(dir_path)
317
318
for filename in files:
319
file_path = f"{dir_path}/{filename}"
320
321
if await ospath.isfile(file_path):
322
size = await ospath.getsize(file_path)
323
print(f"File: {filename} ({size} bytes)")
324
elif await ospath.isdir(file_path):
325
print(f"Directory: {filename}")
326
elif await ospath.islink(file_path):
327
print(f"Link: {filename}")
328
329
# Usage
330
await process_directory('./data')
331
```
332
333
### File Change Detection
334
335
```python
336
import time
337
from aiofiles import ospath
338
339
async def file_changed_since(path: str, timestamp: float) -> bool:
340
"""Check if file was modified since given timestamp."""
341
if not await ospath.exists(path):
342
return False
343
344
if not await ospath.isfile(path):
345
return False
346
347
mtime = await ospath.getmtime(path)
348
return mtime > timestamp
349
350
# Usage
351
last_check = time.time()
352
# ... later ...
353
if await file_changed_since('config.txt', last_check):
354
print("Config file has been modified")
355
```
356
357
## Utility Functions
358
359
```python { .api }
360
def wrap(func):
361
"""
362
Wrap synchronous os.path function to run in executor.
363
364
Parameters:
365
- func: Synchronous os.path function to wrap
366
367
Returns:
368
Async function that runs the original function in thread pool executor
369
"""
370
```
371
372
This utility function can be used to create async versions of other os.path functions not directly provided by aiofiles.
373
374
**Usage Example:**
375
376
```python
377
from aiofiles import ospath
378
import os.path
379
380
# Create async version of os.path.commonpath
381
async_commonpath = ospath.wrap(os.path.commonpath)
382
common = await async_commonpath(['/usr/lib', '/usr/local/lib'])
383
```
384
385
## Error Handling
386
387
Path operations may raise various exceptions:
388
389
- `FileNotFoundError`: Path does not exist
390
- `PermissionError`: Insufficient permissions to access path
391
- `OSError`: General OS-level errors
392
- `ValueError`: Invalid path format
393
394
**Usage Example:**
395
396
```python
397
from aiofiles import ospath
398
399
try:
400
size = await ospath.getsize('nonexistent.txt')
401
except FileNotFoundError:
402
print("File not found")
403
except PermissionError:
404
print("Permission denied")
405
except OSError as e:
406
print(f"OS error: {e}")
407
408
# Safe existence check with error handling
409
async def safe_exists(path: str) -> bool:
410
"""Safely check if path exists."""
411
try:
412
return await ospath.exists(path)
413
except (OSError, ValueError):
414
return False
415
416
exists = await safe_exists('potentially/invalid/path')
417
```