Hướng dẫn python timeout function decorator - Trình trang trí chức năng hết thời gian của python

Tôi có chức năng trang trí sáng tạo hết thời gian sau đây:

class TimeoutError(Exception): pass


def timeout(seconds, error_message = 'Function call timed out'):
    def decorated(func):
        print "timeout: \t" + func.__name__
        def _handle_timeout(signum, frame):
            raise TimeoutError(error_message)


        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, _handle_timeout)
            signal.alarm(seconds)
            try:
                print "timeout wrapper: \t" + func.__name__
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result

        return functools.wraps(func)(wrapper)

    return decorated

Và một người trang trí khác:

import inspect

class withHostAndToken(object):

    __name__ = "withHostAndToken"
    __doc__ = "Get the Host and Token for the API call"

    def __init__(self, func):
        print "withHostAndToken: \t" + func.__name__
        self.func = func
        self.HOST = ''
        self.TOKEN = ''

    def __call__(self,*args, **kwds):

        if self.HOST == '':
            self.HOST = "HOST"
        if self.TOKEN == '':
            self.TOKEN = "TOKEN"

        argsspec = inspect.getargspec(self.func)
        function_args = argsspec[0]
        if 'HOST' in function_args:
            if 'TOKEN' in function_args:
                return self.func(self.HOST , self.TOKEN , *args, **kwds)
            else:
                return self.func(self.HOST  , *args, **kwds)
        elif 'TOKEN' in function_args:
            return self.func(self.TOKEN, *args, **kwds)

Khi tôi cố gắng áp dụng cả hai cho một hàm, tôi không nhận được mã chức năng được gọi:

@timeout(2)
@withHostAndToken
def testDecorators():
    print __name__
    while True:
        print '.'

testDecorators()

Đầu ra của điều này là:

LIÊN QUAN ĐẾN
timeout: withHostAndToken
timeout wrapper: withHostAndToken

Xử lý kết thúc với mã thoát 0

Hiệu suất không đồng bộ

Bộ trang trí thời gian chờ cho các chức năng đồng bộ và không đồng bộ.

Phụ thuộc

Được viết bằng import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 1 thuần túy và không có sự phụ thuộc nào ngoài các thư viện cơ sở.

Cài đặt

pip install .

Từ mã nguồn:

pip install timeout-function-decorator

Từ import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 2:

Nhập khẩu

1. Có ba cách để nhập khẩu trang trí:

from timeout_function_decorator import timeout

1. Nhập trực tiếp từ gói

from timeout_function_decorator.timeout_decorator import timeout

1. Nhập trực tiếp từ mô -đun

from timeout_function_decorator import timeout_decorator

Nhập mô -đun

Trường hợp cuối cùng này rất hữu ích nếu hàm import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 3 đã có trong không gian tên của bạn.

Cách sử dụng

import time
from timeout_function_decorator import timeout


@timeout()
def i_will_never_time_out(value):
    while True:
        time.sleep(1)

        
@timeout(None)
def i_will_never_time_out(value):
    while True:
        time.sleep(1)
        

@timeout(1)
def i_will_not_time_out(value):
    return value


@timeout(1)
def i_will_time_out(value):
    time.sleep(2)
    return value


@timeout(1, RuntimeError)
def i_will_raise_runtime_error(value):
    time.sleep(2)
    return value

Sử dụng người trang trí đơn giản như:

Như bạn có thể nhận thấy, người trang trí yêu cầu dấu ngoặc ngay cả khi không có thông số nào được thông qua.

import asyncio
from timeout_function_decorator import timeout


@timeout()
async def i_will_never_time_out(value):
    while True:
        await asyncio.sleep(1)

        
@timeout(None)
async def i_will_never_time_out(value):
    while True:
        await asyncio.sleep(1)
        

@timeout(1)
async def i_will_not_time_out(value):
    return value


@timeout(1)
async def i_will_time_out(value):
    await asyncio.sleep(2)
    return value


@timeout(1, RuntimeError)
async def i_will_raise_runtime_error(value):
    await asyncio.sleep(2)
    return value

Kết quả tương tự có thể thu được cho các chức năng không đồng bộ:

import inspect

class withHostAndToken(object):

    __name__ = "withHostAndToken"
    __doc__ = "Get the Host and Token for the API call"

    def __init__(self, func):
        print "withHostAndToken: \t" + func.__name__
        self.func = func
        self.HOST = ''
        self.TOKEN = ''

    def __call__(self,*args, **kwds):

        if self.HOST == '':
            self.HOST = "HOST"
        if self.TOKEN == '':
            self.TOKEN = "TOKEN"

        argsspec = inspect.getargspec(self.func)
        function_args = argsspec[0]
        if 'HOST' in function_args:
            if 'TOKEN' in function_args:
                return self.func(self.HOST , self.TOKEN , *args, **kwds)
            else:
                return self.func(self.HOST  , *args, **kwds)
        elif 'TOKEN' in function_args:
            return self.func(self.TOKEN, *args, **kwds)

0

Nếu bạn đã có chức năng

import inspect

class withHostAndToken(object):

    __name__ = "withHostAndToken"
    __doc__ = "Get the Host and Token for the API call"

    def __init__(self, func):
        print "withHostAndToken: \t" + func.__name__
        self.func = func
        self.HOST = ''
        self.TOKEN = ''

    def __call__(self,*args, **kwds):

        if self.HOST == '':
            self.HOST = "HOST"
        if self.TOKEN == '':
            self.TOKEN = "TOKEN"

        argsspec = inspect.getargspec(self.func)
        function_args = argsspec[0]
        if 'HOST' in function_args:
            if 'TOKEN' in function_args:
                return self.func(self.HOST , self.TOKEN , *args, **kwds)
            else:
                return self.func(self.HOST  , *args, **kwds)
        elif 'TOKEN' in function_args:
            return self.func(self.TOKEN, *args, **kwds)

3 trong không gian tên của mình, bạn có thể dễ dàng sử dụng bộ trang trí với không gian tên mô -đun:Warning: When a function times out, an exception is raised but cancellation is not guaranteed. This decorator only notifies the user when enough time has passed since a function call. Handling of the situation and ensuring cancellation is up to the user.

: CẢNH BÁO: CẢNH BÁO: Khi một hàm hết thời gian, một ngoại lệ được nêu ra nhưng hủy bỏ không được đảm bảo. Người trang trí này chỉ thông báo cho người dùng khi đủ thời gian trôi qua kể từ khi gọi chức năng. Xử lý tình huống và đảm bảo hủy bỏ tùy thuộc vào người dùng.

Chữ ký

• Nói chung, người trang trí chấp nhận hai tham số:

• import inspect
  
  class withHostAndToken(object):
  
      __name__ = "withHostAndToken"
      __doc__ = "Get the Host and Token for the API call"
  
      def __init__(self, func):
          print "withHostAndToken: \t" + func.__name__
          self.func = func
          self.HOST = ''
          self.TOKEN = ''
  
      def __call__(self,*args, **kwds):
  
          if self.HOST == '':
              self.HOST = "HOST"
          if self.TOKEN == '':
              self.TOKEN = "TOKEN"
  
          argsspec = inspect.getargspec(self.func)
          function_args = argsspec[0]
          if 'HOST' in function_args:
              if 'TOKEN' in function_args:
                  return self.func(self.HOST , self.TOKEN , *args, **kwds)
              else:
                  return self.func(self.HOST  , *args, **kwds)
          elif 'TOKEN' in function_args:
              return self.func(self.TOKEN, *args, **kwds)
  

  5: A

  import inspect
  
  class withHostAndToken(object):
  
      __name__ = "withHostAndToken"
      __doc__ = "Get the Host and Token for the API call"
  
      def __init__(self, func):
          print "withHostAndToken: \t" + func.__name__
          self.func = func
          self.HOST = ''
          self.TOKEN = ''
  
      def __call__(self,*args, **kwds):
  
          if self.HOST == '':
              self.HOST = "HOST"
          if self.TOKEN == '':
              self.TOKEN = "TOKEN"
  
          argsspec = inspect.getargspec(self.func)
          function_args = argsspec[0]
          if 'HOST' in function_args:
              if 'TOKEN' in function_args:
                  return self.func(self.HOST , self.TOKEN , *args, **kwds)
              else:
                  return self.func(self.HOST  , *args, **kwds)
          elif 'TOKEN' in function_args:
              return self.func(self.TOKEN, *args, **kwds)
  

  6 Chỉ định thời gian chờ tính bằng giây. Nếu

  import inspect
  
  class withHostAndToken(object):
  
      __name__ = "withHostAndToken"
      __doc__ = "Get the Host and Token for the API call"
  
      def __init__(self, func):
          print "withHostAndToken: \t" + func.__name__
          self.func = func
          self.HOST = ''
          self.TOKEN = ''
  
      def __call__(self,*args, **kwds):
  
          if self.HOST == '':
              self.HOST = "HOST"
          if self.TOKEN == '':
              self.TOKEN = "TOKEN"
  
          argsspec = inspect.getargspec(self.func)
          function_args = argsspec[0]
          if 'HOST' in function_args:
              if 'TOKEN' in function_args:
                  return self.func(self.HOST , self.TOKEN , *args, **kwds)
              else:
                  return self.func(self.HOST  , *args, **kwds)
          elif 'TOKEN' in function_args:
              return self.func(self.TOKEN, *args, **kwds)
  

  7, chức năng sẽ không hết thời gian. Mặc định là

  import inspect
  
  class withHostAndToken(object):
  
      __name__ = "withHostAndToken"
      __doc__ = "Get the Host and Token for the API call"
  
      def __init__(self, func):
          print "withHostAndToken: \t" + func.__name__
          self.func = func
          self.HOST = ''
          self.TOKEN = ''
  
      def __call__(self,*args, **kwds):
  
          if self.HOST == '':
              self.HOST = "HOST"
          if self.TOKEN == '':
              self.TOKEN = "TOKEN"
  
          argsspec = inspect.getargspec(self.func)
          function_args = argsspec[0]
          if 'HOST' in function_args:
              if 'TOKEN' in function_args:
                  return self.func(self.HOST , self.TOKEN , *args, **kwds)
              else:
                  return self.func(self.HOST  , *args, **kwds)
          elif 'TOKEN' in function_args:
              return self.func(self.TOKEN, *args, **kwds)
  

  7

import inspect

class withHostAndToken(object):

    __name__ = "withHostAndToken"
    __doc__ = "Get the Host and Token for the API call"

    def __init__(self, func):
        print "withHostAndToken: \t" + func.__name__
        self.func = func
        self.HOST = ''
        self.TOKEN = ''

    def __call__(self,*args, **kwds):

        if self.HOST == '':
            self.HOST = "HOST"
        if self.TOKEN == '':
            self.TOKEN = "TOKEN"

        argsspec = inspect.getargspec(self.func)
        function_args = argsspec[0]
        if 'HOST' in function_args:
            if 'TOKEN' in function_args:
                return self.func(self.HOST , self.TOKEN , *args, **kwds)
            else:
                return self.func(self.HOST  , *args, **kwds)
        elif 'TOKEN' in function_args:
            return self.func(self.TOKEN, *args, **kwds)

9: Loại

@timeout(2)
@withHostAndToken
def testDecorators():
    print __name__
    while True:
        print '.'

testDecorators()

0 được nâng lên. Mặc định là

@timeout(2)
@withHostAndToken
def testDecorators():
    print __name__
    while True:
        print '.'

testDecorators()

1.

Các trường hợp sử dụng dự định

Vỏ bọc sử dụng asyncio và chủ đề để theo dõi thời gian chờ. Điều này thêm một chi phí không tầm thường trên các chức năng được bọc, đặc biệt là trên các chức năng đồng bộ vì chúng cần một luồng để được tạo.

Do đó, các trường hợp sử dụng phổ biến sẽ bao gồm các bộ thử nghiệm trong đó kiểm soát những gì xảy ra khi chúng hết thời gian là quan trọng, vì với các gói như

@timeout(2)
@withHostAndToken
def testDecorators():
    print __name__
    while True:
        print '.'

testDecorators()

2, bạn không thể có các thử nghiệm thất bại với một ngoại lệ được chỉ định.

Mặt khác, các trường hợp sử dụng trong mã sản xuất sẽ được giới hạn trong các chức năng dài với tỷ lệ cuộc gọi thấp vì chi phí là tuyến tính với số lượng cuộc gọi nhưng không phụ thuộc vào kích thước của hàm hoặc thời gian thực hiện của nó.

Một cuốn sổ tay phác thảo tác động của chi phí đối với các hàm đồng bộ hóa và async có thể được tìm thấy ở đây.

Các lô từ máy tính xách tay được trình bày dưới đây để đánh giá hiệu suất nhanh.

Đánh giá hiệu suất

Hiệu suất chung

Hiệu suất đồng bộ hóa

Hiệu suất chức năng nhanh chóng

Hiệu suất chức năng trung bình

Hiệu suất chức năng dài

Hiệu suất chức năng khổng lồ

Hiệu suất không đồng bộ

Hiệu suất chức năng nhanh chóng

Hiệu suất chức năng trung bình

Hiệu suất chức năng dài

Hiệu suất chức năng khổng lồ