UrlTasker (version 0.0.1)
index
urltasker.py

A Python library for defining, templating, and executing configurable
asynchronous actions triggered via URLs.

 
Classes
       
abc.ABC(builtins.object)
Task
AsyncTask
SyncTask
builtins.object
ProcessRequest
ProcessResponse
SocketAsync
TaskFactory
builtins.tuple(builtins.object)
WebScriptsResponse

 
class AsyncTask(Task)
    AsyncTask(code: Union[function, Coroutine], is_awaitable: bool, value: Union[UrlTasker.WebScriptsResponse, PegParser.HttpResponse, UrlTasker.ProcessResponse] = None)
 
This class implements an asynchronous task.
 
 
Method resolution order:
AsyncTask
Task
abc.ABC
builtins.object

Methods defined here:
async run(self) -> None
This method run this asynchronous task.

Data and other attributes defined here:
__abstractmethods__ = frozenset()

Methods inherited from Task:
__init__(self, code: Union[function, Coroutine], is_awaitable: bool, value: Union[UrlTasker.WebScriptsResponse, PegParser.HttpResponse, UrlTasker.ProcessResponse] = None)
Initialize self.  See help(type(self)) for accurate signature.

Data descriptors inherited from Task:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

 
class ProcessRequest(builtins.object)
    ProcessRequest(executable: str, arguments: List[str], environ: Dict[str, str], body: bytes) -> None
 
Dataclass to contains elements about the Process to start it.
 
  Methods defined here:
__eq__(self, other)
Return self==value.
__init__(self, executable: str, arguments: List[str], environ: Dict[str, str], body: bytes) -> None
Initialize self.  See help(type(self)) for accurate signature.
__repr__(self)
Return repr(self).

Data descriptors defined here:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

Data and other attributes defined here:
__annotations__ = {'arguments': typing.List[str], 'body': <class 'bytes'>, 'environ': typing.Dict[str, str], 'executable': <class 'str'>}
__dataclass_fields__ = {'arguments': Field(name='arguments',type=typing.List[str],def...appingproxy({}),kw_only=False,_field_type=_FIELD), 'body': Field(name='body',type=<class 'bytes'>,default=<...appingproxy({}),kw_only=False,_field_type=_FIELD), 'environ': Field(name='environ',type=typing.Dict[str, str],...appingproxy({}),kw_only=False,_field_type=_FIELD), 'executable': Field(name='executable',type=<class 'str'>,defau...appingproxy({}),kw_only=False,_field_type=_FIELD)}
__dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
__hash__ = None
__match_args__ = ('executable', 'arguments', 'environ', 'body')

 
class ProcessResponse(builtins.object)
    ProcessResponse(code: int, body: bytes, error: bytes) -&gt; None
 
Dataclass to contains elements about the process to returns result.
 
  Methods defined here:
__eq__(self, other)
Return self==value.
__init__(self, code: int, body: bytes, error: bytes) -> None
Initialize self.  See help(type(self)) for accurate signature.
__repr__(self)
Return repr(self).

Data descriptors defined here:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

Data and other attributes defined here:
__annotations__ = {'body': <class 'bytes'>, 'code': <class 'int'>, 'error': <class 'bytes'>}
__dataclass_fields__ = {'body': Field(name='body',type=<class 'bytes'>,default=<...appingproxy({}),kw_only=False,_field_type=_FIELD), 'code': Field(name='code',type=<class 'int'>,default=<da...appingproxy({}),kw_only=False,_field_type=_FIELD), 'error': Field(name='error',type=<class 'bytes'>,default=...appingproxy({}),kw_only=False,_field_type=_FIELD)}
__dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
__hash__ = None
__match_args__ = ('code', 'body', 'error')

 
class SocketAsync(builtins.object)
     Methods defined here:
__init__(self)
Initialize self.  See help(type(self)) for accurate signature.
async connect(self, destination: Tuple[str, int]) -> None
This method connect the async socket.
async recv(self) -> bytearray
This method receives encrypted data from server with SSL encryption.
async sendall(self, data: bytes = None) -> None
This methods sends data to server with SSL encryption.
async start_ssl(self, insecure: bool = False) -> None
This method starts the TLS connection.

Data descriptors defined here:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

 
class SyncTask(Task)
    SyncTask(code: Union[function, Coroutine], is_awaitable: bool, value: Union[UrlTasker.WebScriptsResponse, PegParser.HttpResponse, UrlTasker.ProcessResponse] = None)
 
This class implements a synchronous task.
 
 
Method resolution order:
SyncTask
Task
abc.ABC
builtins.object

Methods defined here:
run(self) -> None
This method run this synchronous task.

Data and other attributes defined here:
__abstractmethods__ = frozenset()

Methods inherited from Task:
__init__(self, code: Union[function, Coroutine], is_awaitable: bool, value: Union[UrlTasker.WebScriptsResponse, PegParser.HttpResponse, UrlTasker.ProcessResponse] = None)
Initialize self.  See help(type(self)) for accurate signature.

Data descriptors inherited from Task:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

 
class Task(abc.ABC)
    Task(code: Union[function, Coroutine], is_awaitable: bool, value: Union[UrlTasker.WebScriptsResponse, PegParser.HttpResponse, UrlTasker.ProcessResponse] = None)
 
This class implements the abstract class for tasks.
 
 
Method resolution order:
Task
abc.ABC
builtins.object

Methods defined here:
__init__(self, code: Union[function, Coroutine], is_awaitable: bool, value: Union[UrlTasker.WebScriptsResponse, PegParser.HttpResponse, UrlTasker.ProcessResponse] = None)
Initialize self.  See help(type(self)) for accurate signature.
run(self) -> None

Data descriptors defined here:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

Data and other attributes defined here:
__abstractmethods__ = frozenset({'run'})

 
class TaskFactory(builtins.object)
    This class implements the task builder.
 
  Static methods defined here:
create(code: Union[function, Coroutine]) -> UrlTasker.Task
This method implements builts the task.

Data descriptors defined here:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

 
class WebScriptsResponse(builtins.tuple)
    WebScriptsResponse(body, error, code)
 
WebScriptsResponse(body, error, code)
 
 
Method resolution order:
WebScriptsResponse
builtins.tuple
builtins.object

Methods defined here:
__getnewargs__(self)
Return self as a plain tuple.  Used by copy and pickle.
__repr__(self)
Return a nicely formatted representation string
_asdict(self)
Return a new dict which maps field names to their values.
_replace(self, /, **kwds)
Return a new WebScriptsResponse object replacing specified fields with new values

Class methods defined here:
_make(iterable) from builtins.type
Make a new WebScriptsResponse object from a sequence or iterable

Static methods defined here:
__new__(_cls, body, error, code)
Create new instance of WebScriptsResponse(body, error, code)

Data descriptors defined here:
body
Alias for field number 0
error
Alias for field number 1
code
Alias for field number 2

Data and other attributes defined here:
__match_args__ = ('body', 'error', 'code')
_field_defaults = {}
_fields = ('body', 'error', 'code')

Methods inherited from builtins.tuple:
__add__(self, value, /)
Return self+value.
__contains__(self, key, /)
Return key in self.
__eq__(self, value, /)
Return self==value.
__ge__(self, value, /)
Return self>=value.
__getattribute__(self, name, /)
Return getattr(self, name).
__getitem__(self, key, /)
Return self[key].
__gt__(self, value, /)
Return self>value.
__hash__(self, /)
Return hash(self).
__iter__(self, /)
Implement iter(self).
__le__(self, value, /)
Return self<=value.
__len__(self, /)
Return len(self).
__lt__(self, value, /)
Return self<value.
__mul__(self, value, /)
Return self*value.
__ne__(self, value, /)
Return self!=value.
__rmul__(self, value, /)
Return value*self.
count(self, value, /)
Return number of occurrences of value.
index(self, value, start=0, stop=9223372036854775807, /)
Return first index of value.
 
Raises ValueError if the value is not present.

Class methods inherited from builtins.tuple:
__class_getitem__(...) from builtins.type
See PEP 585

 
Functions
       
exit(status=None, /)
Exit the interpreter by raising SystemExit(status).
 
If the status is omitted or None, it defaults to zero (i.e., success).
If the status is an integer, it will be used as the system exit status.
If it is another kind of object, it will be printed and the system
exit status will be one (i.e., failure).
get_running_loop()
Return the running event loop.  Raise a RuntimeError if there is none.
 
This function is thread-specific.
get_task(url: str, **kwargs: Dict[str, str]) -> UrlTasker.Task
This function returns a code (function or coroutine) to run url as task.
async handle_executable(url: Dict[str, List[bytearray]], **kwargs) -> UrlTasker.ProcessResponse
Run a subprocess asynchronously using full path and argument list.
Sends input_data to STDIN (if provided), and captures STDOUT and STDERR.
async handle_http(url: Dict[str, List[bytearray]], method: str = 'GET', body: bytes = b'', headers: List[Tuple[str, str]] = [], version: float = 1.0, insecure: bool = False, **kwargs) -> PegParser.HttpResponse
This function is the asynchronous HTTP handle to perform request
and receive response.
async handle_method_http_body(url: Dict[str, List[bytearray]], **kwargs) -> PegParser.HttpResponse
This function is the asynchronous HTTP handle to perform complete request
and receive response.
async handle_python(url: Dict[str, List[bytearray]], **kwargs) -> UrlTasker.ProcessResponse
Run a subprocess asynchronously using full path and argument list.
Sends input_data to STDIN (if provided), and captures STDOUT and STDERR.
async handle_webscripts(url: Dict[str, List[bytearray]], **kwargs) -> UrlTasker.WebScriptsResponse
This function sends an async WebScripts requests from an URL.
async http_request(request: PegParser.HttpRequest, port: int, use_ssl: bool, insecure: bool, **kwargs) -> PegParser.HttpResponse
Perform an async HTTP request and return the parsed HTTP response.
This version avoids Windows IOCP timeouts by using loop.sock_recv().
main() -> int
This function is the main function to start the script
from the command line.
async run_executable(request: UrlTasker.ProcessRequest, **kwargs) -> UrlTasker.ProcessResponse
Run a subprocess asynchronously using full path and argument list.
Sends input_data to STDIN (if provided), and captures STDOUT and STDERR.
async run_tasks(*urls: List[str]) -> List[UrlTasker.Task]
This function run tasks from URLs.

 
Data
        AF_INET = <AddressFamily.AF_INET: 2>
Coroutine = typing.Coroutine
Dict = typing.Dict
List = typing.List
PIPE = -1
SOCK_STREAM = <SocketKind.SOCK_STREAM: 1>
Tuple = typing.Tuple
Union = typing.Union
__author_email__ = 'mauricelambert434@gmail.com'
__copyright__ = '\nUrlTasker Copyright (C) 2025 Maurice Lambert\n...ome to redistribute it\nunder certain conditions.\n'
__description__ = '\nA Python library for defining, templating, and ...gurable\nasynchronous actions triggered via URLs.\n'
__license__ = 'GPL-3.0 License'
__maintainer__ = 'Maurice Lambert'
__maintainer_email__ = 'mauricelambert434@gmail.com'
__url__ = 'https://github.com/mauricelambert/UrlTasker'
copyright = '\nUrlTasker Copyright (C) 2025 Maurice Lambert\n...ome to redistribute it\nunder certain conditions.\n'
environ = environ({'ALLUSERSPROFILE': 'C:\\ProgramData', '...INDIR': 'C:\\WINDOWS', 'ZES_ENABLE_SYSMAN': '1'})
executable = r'C:\Program Files\Python310\python.exe'
handlers = {'exe': <function handle_executable>, 'executable': <function handle_executable>, 'http': <function handle_http>, 'http+CONNECT': <function handle_method_http_body>, 'http+DELETE': <function handle_method_http_body>, 'http+GET': <function handle_method_http_body>, 'http+HEAD': <function handle_method_http_body>, 'http+OPTIONS': <function handle_method_http_body>, 'http+PATCH': <function handle_method_http_body>, 'http+POST': <function handle_method_http_body>, ...}
license = 'GPL-3.0 License'

 
Author
        Maurice Lambert