Một tình huống phổ biến đối với các tác giả thư viện là họ chấp nhận một số khả năng gọi được dưới dạng gọi lại cho logic do người dùng xác định
Nếu tác giả thư viện muốn thêm hỗ trợ cho các phương thức không đồng bộ, thì thường cần có một số thay đổi cấp cao, nhưng có một vấn đề dẫn đến việc thấm qua tất cả các loại chức năng tiện ích
Thay vì các ví dụ về đồ chơi, tôi sẽ sử dụng một số mã mà tôi đang làm việc trong một nhánh của thư viện
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
def _load_location_data[self, *, schema, req, location]:
loader_func = self._get_loader[location]
return loader_func[req, schema]
async def _async_load_location_data[self, *, schema, req, location]:
loader_func = self._get_loader[location]
if asyncio.iscoroutinefunction[loader_func]:
data = await loader_func[req, schema]
else:
data = loader_func[req, schema]
return data
Cả hai chức năng này đều “giống nhau”, nhưng chúng ta cần cả hai. Điều này không quá tệ đối với một chức năng đơn lẻ, nhưng hãy sắp xếp một số hook và phương thức riêng biệt, và cuối cùng, bạn sẽ tăng gấp đôi kích thước của rất nhiều hệ thống ống nước trong dự án một cách hiệu quả để cho phép một đường dẫn cuộc gọi hoàn toàn không đồng bộ cùng với một đường dẫn đồng bộ hóa.
Giải pháp hiện tại cho người trang trí
Giao diện cung cấp cho người dùng đôi khi cần thể hiện sự khác biệt giữa hai phiên bản của cùng một mã, đ. g.
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
Nhưng chúng ta có thể ẩn nó rất nhiều lần bằng cách sử dụng công cụ trang trí và kiểm tra nhanh.
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
Điều này hoạt động tuyệt vời cho các trường hợp như
import flask
from webargs.flaskparser import parser
app = flask.Flask[__name__]
@app.route["/foo"]
@parser.use_args[...]
async def foo[...]: ...
Do đó, tôi không quá quan tâm đến việc cố gắng tìm cách trình bày giao diện tốt hơn để người dùng gọi các biến thể đồng bộ hóa hoặc không đồng bộ của mã thư viện. Trang trí giải quyết vấn đề này khá tốt, nơi chúng ta có thể sử dụng chúng. Và bạn có thể hỗ trợ
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
Các cuộc thảo luận trước đây
Vấn đề này đã được thảo luận trước đây, đặc biệt
- Có cách nào để xác định chức năng có thể được sử dụng làm chức năng bình thường và chức năng không đồng bộ không?
- Gói các chức năng không đồng bộ để sử dụng trong mã đồng bộ hóa - #14 của TobiasHT
cả hai dường như có liên quan
Tuy nhiên, tôi không thấy ai yêu cầu điều gì – với tư cách là tác giả thư viện – có vẻ như là giải pháp tốt nhất.
Có cách nào để thay đổi ngôn ngữ sao cho việc xây dựng các biến thể không đồng bộ và không đồng bộ của cùng chức năng có thể được tự động hóa hoặc đơn giản hóa không?
Nếu có một chủ đề quá khứ khác tôi nên đọc, xin vui lòng cho tôi biết
giải pháp lý tưởng
Hôm nay tôi có cái này
________số 8và những gì tôi muốn viết thay vào đó là điều này
class Parser:
maybe_async def _load_location_data[self, *, schema, req, location]:
loader_func = self._get_loader[location]
if [
asyncio.iscoroutinefunction[loader_func] and
MAGIC_is_currently_async
]:
data = await loader_func[req, schema]
else:
data = loader_func[req, schema]
return data
maybe_async def _other_helper_func[self, ...]:
# other magic -- strip the await in synchronous calls
return await self._load_location_data[...]
def public_func[self, ...]:
return self._other_helper_func[...]
async def async_public_func[self, ...]:
return await self._other_helper_func.call_async[...]
# why limit 'maybe_async' to internal methods?
# if it's part of the lanaguage, we also get to avoid the split in public
maybe_async def alternative_public_func[self, ...]: ...
Tôi biết rằng một số điều này có thể được thực hiện bằng cách tạo mã. Tuy nhiên, việc duy trì mã nguồn
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
Kết luận và câu hỏi cuối cùng
Có giải pháp nào có thể được viết để thực hiện những điều trên [rõ ràng là có ít đường cú pháp hơn] trong ngôn ngữ ngày nay không?
Mục tiêu là cải thiện việc bảo trì thư viện. Vì vậy, việc thêm các phụ thuộc thời gian chạy vào các gói pypi khác hoặc các giải pháp rất phức tạp không thực sự giải quyết được vấn đề
Có kỹ thuật nào đã biết để thực hiện chia sẻ mã giữa hai đường dẫn giúp vấn đề này bớt nghiêm trọng hơn không?
Tôi không ngạc nhiên; . Nhưng nếu mọi người đang tìm cách riêng để đạt được điều này ngày hôm nay, liệu có khả năng đưa một trong những quy ước + người trợ giúp đó vào stdlib không?
Sau đây là một giải pháp tốt như bất kỳ giải pháp nào khác nếu nó hoạt động
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
Nhưng tôi không biết một cách để làm điều đó. Tạo biến thể đồng bộ hóa từ chức năng không đồng bộ là một bước tốt, nhưng đó chỉ là một phần của vấn đề. Nếu chúng ta có
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
và
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
Tôi vô cùng biết ơn các liên kết đến bất kỳ công cụ hiện có nào thực hiện việc này, chỉ với mục đích học hỏi. Tôi đã xem xét asgiref một chút, nhưng có vẻ như nó chủ yếu bao gồm các cuộc gọi trong chuỗi nền
Có thể không có nhiều người được hưởng lợi từ việc bổ sung
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
Về chủ đề khác, tôi không thể chuyển chủ đề này sang async-sig. Tôi nhận được "Bạn không được phép xem tài nguyên được yêu cầu. ” Có lẽ tôi không được phép đăng ở đó?
Cảm ơn vì đã chuyển cái này sang async-sig.
Về vấn đề chính, tôi cảm thấy mình phải nhắc lại - tốt hơn hết là bạn nên phát minh ra giải pháp của riêng mình phù hợp với khuôn khổ
Tôi không muốn tạo ấn tượng rằng tôi không lắng nghe và tôi xin lỗi nếu tôi nói bất cứ điều gì gợi ý rằng.
Tôi đang cố gắng tìm hiểu cách tốt nhất để xử lý tình huống này là gì. Và tôi muốn hệ thống hóa điều đó - có lẽ trong tài liệu asyncio hoặc ở nơi nào khác phù hợp - để bất kỳ ai khác đang cố gắng làm những điều tương tự đều có sẵn phương pháp hay nhất đó.
Nếu giải pháp đơn giản như hàm 6 dòng, thì không có lý do gì để không thêm nó vào
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
Nếu không có gì khác, chúng ta phải lo lắng về việc người gọi đã có một vòng lặp đang chạy. asgiref sử dụng một luồng nền để chạy một vòng lặp riêng biệt và đồng hồ ở khoảng 200 LỘC. Nhóm đó đã tìm ra giải pháp thay thế an toàn và đáng tin cậy cho hầu hết các trường hợp chưa? . không thể hy vọng
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper
def decorator[func]:
if asyncio.iscoroutinefunction[func]:
@functools.wraps[func]
async def wrapper[*args, **kwargs]: ...
else:
@functools.wraps[func]
def wrapper[*args, **kwargs]: ...
return wrapper