o Oj;!@s|ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z ddl m Z ddlmZddlmZddlmZddlmZddlmZmZddlZd d lmZd d lmZmZmZm Z erpdd lm!Z!d Z"e#dZ$ddddedddddd ddd dee%Bde%e defBde&edfde'e%efdBddde e(egdfdBde ee%ge)fdBd e*d!e+d"e+d#e)dBd$e+d%e+d&e)d'e)d(e+f d)d*Z,ddddeddddddd+ dee%Bde%e defBde&edfde'e%efdBddde e(egefdBde ee%ge)fdBd e*d!e+d"e+d#e)dBd&e)d'e)d(e+fd,d-Z-ed.Z.d/e%d(e/e%fd0d1Z0 dLde%e defBdd2de&edfde'e%efdBd3e(edBd(d4f d5d6Z1de%e defBd(d2fd7d8Z2Gd9d4d4Z3d:e%d;e%dBde&edfde'e%efd(df de%d(efd?d@Z5d(e%dBfdAdBZ6ej7d;e%dBd(e dCfdDdEZ8dFe+dGed(dfdHdIZ9dMdJdKZ:dS)NN)Callable Generator) import_module) get_context) SpawnProcess)Path)sleep) TYPE_CHECKINGAny) DefaultFilter)Change FileChangeawatchwatch)Literal) run_process arun_processdetect_target_type import_stringzwatchfiles.mainautoi@2TF) argskwargs target_typecallback watch_filter grace_perioddebouncestepdebugsigint_timeoutsigkill_timeout recursiveignore_permission_deniedpathstarget.rrrz&Literal['function', 'command', 'auto']rrrr r!r"r#r$r%r&returnc Gs|dkrt|}td||tt||||}d}|r'td|t|z0t||||| d| | dD]}|o<|||j| | dt|||||}|d7}q5W||S|w) u Run a process and restart it upon file changes. `run_process` can work in two ways: * Using `multiprocessing.Process` † to run a python function * Or, using `subprocess.Popen` to run a command !!! note **†** technically `multiprocessing.get_context('spawn').Process` to avoid forking and improve code reload/import. Internally, `run_process` uses [`watch`][watchfiles.watch] with `raise_interrupt=False` so the function exits cleanly upon `Ctrl+C`. Args: *paths: matches the same argument of [`watch`][watchfiles.watch] target: function or command to run args: arguments to pass to `target`, only used if `target` is a function kwargs: keyword arguments to pass to `target`, only used if `target` is a function target_type: type of target. Can be `'function'`, `'command'`, or `'auto'` in which case [`detect_target_type`][watchfiles.run.detect_target_type] is used to determine the type. callback: function to call on each reload, the function should accept a set of changes as the sole argument watch_filter: matches the same argument of [`watch`][watchfiles.watch] grace_period: number of seconds after the process is started before watching for changes debounce: matches the same argument of [`watch`][watchfiles.watch] step: matches the same argument of [`watch`][watchfiles.watch] debug: matches the same argument of [`watch`][watchfiles.watch] sigint_timeout: the number of seconds to wait after sending sigint before sending sigkill sigkill_timeout: the number of seconds to wait after sending sigkill before raising an exception recursive: matches the same argument of [`watch`][watchfiles.watch] Returns: number of times the function was reloaded. ```py title="Example of run_process running a function" from watchfiles import run_process def callback(changes): print('changes detected:', changes) def foobar(a, b): print('foobar called with:', a, b) if __name__ == '__main__': run_process('./path/to/dir', target=foobar, args=(1, 2), callback=callback) ``` As well as using a `callback` function, changes can be accessed from within the target function, using the `WATCHFILES_CHANGES` environment variable. ```py title="Example of run_process accessing changes" from watchfiles import run_process def foobar(a, b, c): # changes will be an empty list "[]" the first time the function is called changes = os.getenv('WATCHFILES_CHANGES') changes = json.loads(changes) print('foobar called due to changes:', changes) if __name__ == '__main__': run_process('./path/to/dir', target=foobar, args=(1, 2, 3)) ``` Again with the target as `command`, `WATCHFILES_CHANGES` can be used to access changes. ```bash title="example.sh" echo "changers: ${WATCHFILES_CHANGES}" ``` ```py title="Example of run_process running a command" from watchfiles import run_process if __name__ == '__main__': run_process('.', target='./example.sh') ``` rrunning "%s" as %sr3sleeping for %s seconds before watching for changesF)rr r!r"raise_interruptr%r&)r#r$r )rloggerr" catch_sigterm start_processrrstop)r(rrrrrrr r!r"r#r$r%r&r'processreloadschangesrrT/var/www/html/pharmsmart-cdr-gen/venv/lib/python3.10/site-packages/watchfiles/run.pyrs:`    r) rrrrrrr r!r"r%r&c  sddl} |dkr t|}td||ttjt||||IdH}d}|r6td|t |IdHt | |||| | | d2z43dHW}|durZ||}| |rZ|IdHtj|j IdHtjt|||||IdH}|d7}qB6tj|j IdH|S)a  Async equivalent of [`run_process`][watchfiles.run_process], all arguments match those of `run_process` except `callback` which can be a coroutine. Starting and stopping the process and watching for changes is done in a separate thread. As with `run_process`, internally `arun_process` uses [`awatch`][watchfiles.awatch], however `KeyboardInterrupt` cannot be caught and suppressed in `awatch` so these errors need to be caught separately, see below. ```py title="Example of arun_process usage" import asyncio from watchfiles import arun_process async def callback(changes): await asyncio.sleep(0.1) print('changes detected:', changes) def foobar(a, b): print('foobar called with:', a, b) async def main(): await arun_process('.', target=foobar, args=(1, 2), callback=callback) if __name__ == '__main__': try: asyncio.run(main()) except KeyboardInterrupt: print('stopped via KeyboardInterrupt') ``` rNrr*r+)rr r!r"r%r&r ) inspectrr-r"r.anyio to_threadrun_syncr/rr isawaitabler0)r(rrrrrrr r!r"r%r&r'r5r1r2r3rrrr4rs<-     rspawncmdcCs(ddl}|jdk}tj||dS)Nrwindows)posix)platformunamesystemlowershlexsplit)r<r?r>rrr4 split_cmdsrEzLiteral['function', 'command']r3CombinedProcessc Cs|durd}n tdd|D}|tjd<|dkrA|pi}t|tr/|t||f}t}i}n|}tj |||d}| t|S|sE|rJt dt|tsSJdt |}t|}t|S) Nz[]cSsg|] \}}||gqSr)raw_str).0cprrr4 sz!start_process..WATCHFILES_CHANGESfunction)r(rrz-ignoring args and kwargs for "command" targetz+target must be a string to run as a command)jsondumpsosenviron isinstancestr get_tty_path run_function spawn_contextProcessstartr-warningrE subprocessPopenrF) r(rrrr3changes_env_vartarget_r1 popen_argsrrr4r/s(    r/cCs0t|tsdS|drdStd|rdSdS)a^ Used by [`run_process`][watchfiles.run_process], [`arun_process`][watchfiles.arun_process] and indirectly the CLI to determine the target type with `target_type` is `auto`. Detects the target type - either `function` or `command`. This method is only called with `target_type='auto'`. The following logic is employed: * If `target` is not a string, it is assumed to be a function * If `target` ends with `.py` or `.sh`, it is assumed to be a command * Otherwise, the target is assumed to be a function if it matches the regex `[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)+` If this logic does not work for you, specify the target type explicitly using the `target_type` function argument or `--target-type` command line argument. Args: target: The target value Returns: either `'function'` or `'command'` rM)z.pyz.shcommandz[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)+)rRrSendswithre fullmatch)r(rrr4rs   rc@sveZdZdddZddeded d fd d Zd efd dZed efddZ ded d fddZ ed ed BfddZ d S)rFrJ&SpawnProcess | subprocess.Popen[bytes]cCs||_|jdus JddS)Nzprocess not yet spawned_ppid)selfrJrrr4__init__?szCombinedProcess.__init__rr r#r$r)NcCstjdd|rPtdt|jtj z| |Wnt j y/t d|Ynw|jdurIt dt|jtj| |dStddSt d|jdS)NrLzstopping process...z!SIGINT timed out after %r secondsz+process has not terminated, sending SIGKILLzprocess stoppedz#process already dead, exit code: %d)rPrQpopis_aliver-r"killrfsignalSIGINTjoinrZTimeoutExpiredrYexitcodeSIGKILL)rgr#r$rrr4r0Cs     zCombinedProcess.stopcCs$t|jtr |jS|jduSN)rRrerrjpollrgrrr4rj[s  zCombinedProcess.is_alivecCs|jjSrrrdrtrrr4rfaszCombinedProcess.pidtimeoutcCs,t|jtr|j|dS|j|dSrr)rRrerrnwait)rgrurrr4rnfs zCombinedProcess.joincCst|jtr |jjS|jjSrr)rRrerrp returncodertrrr4rpls zCombinedProcess.exitcode)rJrc)rr ) __name__ __module__ __qualname__rhintr0boolrjpropertyrfrnrprrrr4rF>s rMtty_pathcCsDt|t|}||i|WddS1swYdSrr)set_ttyr)rMr~rrfuncrrr4rUts "rU dotted_pathc Csz |ddd\}}Wnty"}z td|d|d}~wwt|}zt||WStyE}z td|d|d |d}~ww) z Stolen approximately from django. Import a dotted module path and return the attribute/class designated by the last name in the path. Raise ImportError if the import fails.  .r "z!" doesn't look like a module pathNzModule "z" does not define a "z " attribute)striprsplit ValueError ImportErrorrgetattrAttributeError)r module_path class_nameemodulerrr4rzs rcCs:z ttjWStyYdStyYdSw)zr Return the path to the current TTY, if any. Virtually impossible to test in pytest, hence no cover. z/dev/ttyN)rPttynamesysstdinfilenoOSErrorrrrrr4rTs  rT)NNNccsj|r0zt|}|t_dVWdWdS1swYWdSty/dVYdSwdVdSrr)openrrr)r~ttyrrr4rs &   rsignum_framecCstdt|t)Nz-received signal %s, raising KeyboardInterrupt)r-rYrlSignalsKeyboardInterrupt)rrrrr4raise_keyboard_interruptsrcCs"tdtttjtdS)a Catch SIGTERM and raise KeyboardInterrupt instead. This means watchfiles will stop quickly on `docker compose stop` and other cases where SIGTERM is sent. Without this the watchfiles process will be killed while a running process will continue uninterrupted. z8registering handler for SIGTERM on watchfiles process %dN)r-r"rPgetpidrlSIGTERMrrrrr4r.sr.rr)r)N); contextlibrNloggingrPrarCrlrZrcollections.abcrr importlibrmultiprocessingrmultiprocessing.contextrpathlibrtimertypingr r r6filtersr mainr rrrr__all__ getLoggerr-rStupledictsetr|floatr{rrrVlistrEr/rrFrUrrTcontextmanagerrrr.rrrr4s                      R    $ 26