>> coalesce(None, 1) 1 >>> coalesce() >>> coalesce(0, False, True) 0 >>> coalesce(0, False, True, test_value=0) False # This won't work because of the `is not test_value` type testing: >>> coalesce([], dict(spam='eggs'), test_value=[]) [] c34># UH oTLdM Uv M g7fN).0argrs K/home/james-whalen/.local/lib/python3.13/site-packages/portalocker/utils.py coalesce..5s>:(=s N)next)rargss` rcoalescers, >> EEc## [U[R5(aUnO[R"U5nUR5(a U<S35eURR SSS9 [ R"U=(a S=(d S[UR5SS9nUv UR5 [R"UR55 SSS5 [R"WRU5 [R "["5 [R$"UR5 SSS5 g!,(df  Nt=f!,(df  g=f![R "["5 [R$"WR5 SSS5 f!,(df  f=f=f7f) aOpen a file for atomic writing. Instead of locking this method allows you to write the entire file and move it to the actual location. Note that this makes the assumption that a rename is atomic on your platform which is generally the case but not a guarantee. http://docs.python.org/library/os.html#os.rename >>> filename = 'test_file.txt' >>> if os.path.exists(filename): ... os.remove(filename) >>> with open_atomic(filename) as fh: ... written = fh.write(b'test') >>> assert os.path.exists(filename) >>> os.remove(filename) >>> import pathlib >>> path_filename = pathlib.Path('test_file.txt') >>> with open_atomic(path_filename) as fh: ... written = fh.write(b'test') >>> assert path_filename.exists() >>> path_filename.unlink() z existsT)parentsexist_okwbwF)modedirdeleteN) isinstancepathlibPathexistsparentmkdirtempfileNamedTemporaryFilestrflushosfsyncfilenorenamename contextlibsuppress Exceptionremove)filenamebinarypathtemp_fhs rr r 8s.>(GLL))||H%{{}}000  KKdT2  $ $o %#       !" $ ',,%   + IIgll #, +  , +Z  + IIgll #, + +sfB.G09E)G2!E9G-!E( G E%!G( E62G9G!F>5 G> G GGc\rSrSr%S\S'S\S'S\S'SSSjjr\RSSS jj5rSS jr \RSS j5r SS jr SS jr SSjr Srg)LockBaseqfloattimeoutcheck_intervalboolfail_when_lockedNc[U[5Ul[U[5Ul[U[ 5Ulgr)rDEFAULT_TIMEOUTr?DEFAULT_CHECK_INTERVALr@DEFAULT_FAIL_WHEN_LOCKEDrBselfr?r@rBs r__init__LockBase.__init__ys5  9 &~7MN (  $! rcgrrrGs racquireLockBase.acquires $'rc## [XRS5n[X RS5nSv Sn[R"5nXc-[R"5:adUS- nUv [R"5U- n[R "[ SXT-U- 55 Xc-[R"5:aMcgg7f)NgrrgMbP?)rr?r@time perf_countersleepmax)rHr?r@ f_timeoutf_check_intervali start_timesince_start_times r_timeout_generatorLockBase._timeout_generators WllC8 #N4G4GM &&( $t'8'8':: FAG $002Z?  JJs51#7;K"KL M $t'8'8'::s CC C cgrrrHs rreleaseLockBase.releases"rc"UR5$rrLr[s r __enter__LockBase.__enter__||~rc$UR5 grr\)rHexc_type exc_value tracebacks r__exit__LockBase.__exit__s rc$UR5 grrd)rHinstances r __delete__LockBase.__delete__sr)r@rBr?NNN)r? float | Noner@rorB bool | NonereturnNoner?ror@rorBrprqtyping.IO[typing.AnyStr])r?ror@rorqztyping.Iterator[int]rqrrrqrt)reztype[BaseException] | NonerfzBaseException | Nonerg typing.Anyrqrp)rkr<rqrr)__name__ __module__ __qualname____firstlineno____annotations__rIabcabstractmethodrLrXr\r`rhrl__static_attributes__rrrr<r<qs N!%'+(,    %  &     !%'+(, ''%'& ' " ''NN%N  N( "",(   rr<c ^\rSrSr%SrS\S'S\S'S\S'S\S 'S \S 'S \S 'S\S 'S\S'S\S'SS\\\4SU4Sjjjr SSSjjr SSjr S Sjr S!Sjr S"SjrS"SjrSrU=r$)#r a7Lock manager with built-in timeout Args: filename: filename mode: the open mode, 'a' or 'ab' should be used for writing. When mode contains `w` the file will be truncated to 0 bytes. timeout: timeout when trying to acquire a lock check_interval: check interval while waiting fail_when_locked: after the initial lock failed, return an error or lock the file. This does not wait for the timeout. **file_open_kwargs: The kwargs for the `open(...)` call fail_when_locked is useful when multiple threads/processes can race when creating a file. If set to true than the system will wait till the lock was acquired and then return an AlreadyLocked exception. Note that the file is opened first and locked later. So using 'w' as mode will result in truncate _BEFORE_ the lock is checked. ztypes.IO | Nonefhr,r7r!rAtruncater>r?r@rBconstants.LockFlagsflagszdict[str, typing.Any]file_open_kwargsaNc r>SU;a.Sn[R"[URSS55nOSnUc[nO7U[ R R-(d[R"SSS9 SUl [U5Ul X l XlX`lXpl["T U]IX4U5 g)Nr TrF&timeout has no effect in blocking moder stacklevel)typingcastr replacerDr LockFlags NON_BLOCKINGwarningswarnrr,r7r!rrrsuperrI) rHr7r!r?r@rBrrr __class__s rrI Lock.__init__s $;H;;tT\\#s%;Tb4[R"[5 TR5 SSS5 gg!,(df  g=fr)r3r4r5close)rsr try_closeLock.acquire..try_closes3~((3HHJ4333s : Aru)rrBrrrrrrrrrIOAnyStr_get_fhrX _get_lockr LockException AlreadyLockedr5 _prepare_fh) rHr?r@rBr exception_excrs @rrL Lock.acquiresl$$46K6KLi11>>># MM8  WW ;;vyy7< <\\^   ((AAI =^^B' B0  KO  b !{{699V]]3R887++ G $K$229=3F$ =  ..s3<  =s$EF9'F  F9F44F9c"UR5$rr_r[s rr`Lock.__enter__;rbrcUR(aB[R"UR5 URR5 SUlgg)z)Releases the currently locked file handleN)rrunlockrr[s rr\ Lock.release>s4 77   tww ' GGMMODG rc[R"[R[ UR UR 40URD65$)zGet a new filehandle)rrrropenr7r!rr[s rr Lock._get_fhEs@{{ HH    ''   rcF[R"XR5 U$)zD Try to lock the given filehandle returns LockException if it fails)rlockrrHrs rrLock._get_lockPs ZZ( rclUR(a"URS5 URS5 U$)zo Prepare the filehandle for usage If truncate is a number, the file will be truncated to that amount of bytes r)rseekrs rrLock._prepare_fhXs& == GGAJ KKN r)rrr7rr!r)r7r r!r r?ror@r>rBrArrrrwrqrrrnrsrvru)rqtypes.IO)rrrqr)rxryrzr{__doc__r|rErF LOCK_METHODrIrLr`r\rrrr __classcell__rs@rr r s ( M IN N ++  $ 6!9%0DDD D  D  D#D'D DDD!%'+(, E9E9%E9& E9 " E9N    rc^\rSrSrSrS\\S\4S U4SjjjrS S U4Sjjjr S U4Sjjr Sr U=r $) RLockifz A reentrant lock, functions in a similar way to threading.RLock in that it can be acquired multiple times. When the corresponding number of release() calls are made the lock will finally release the underlying file lock. rFc<>[TU]UUUUUU5 SUlg)Nr)rrI_acquire_count)rHr7r!r?r@rBrrs rrIRLock.__init__ms.          rc>URS:aA[R"[R[RUR 5nO[ TU]XU5nU=RS- slUceU$)Nr)rrrrrrrrL)rHr?r@rBrrs rrL RLock.acquirese   ! #VYYv}}5tww?B:JKB q ~~ rc>URS:Xa[R"S5eURS:Xa[TU]5 U=RS-slg)Nrz'Cannot release more times than acquiredr)rrrrr\)rHrs rr\ RLock.releasesR   ! #**9    ! # GO  q r)r)r7r r!r r?r>r@r>rBrArrrqrrrnrsru) rxryrzr{rrDrErrIrLr\rrrs@rrrfs( 6!&%0         #    *!%'+(,  % &  "  !!rrcZ^\rSrSrS\\S\4SU4SjjjrSSjrSr U=r $) TemporaryFileLockiz.lockTc j>[TU]USUUUUS9 [R"UR5 g)Nr )r7r!r?r@rBr)rrIatexitregisterr\)rHr7r?r@rBrrs rrITemporaryFileLock.__init__s; )-    %rc[RU5 [RR UR 5(a![R "UR 5 ggr)r r\r.r9isfiler7unlinkr[s rr\TemporaryFileLock.releases: T 77>>$-- ( ( IIdmm $ )rr) r7r,r?r>r@r>rBrArrrqrrru) rxryrzr{rDrErrIr\rrrs@rrrsb ( 6!%%0 &&& &  & # & &&$%%rrc^\rSrSr%SrS\S'SS\R"5\\ S4SU4Sjjjr SS jr SS jr SS jr SSS jjrSS jrSSjrSrU=r$)BoundedSemaphoreia Bounded semaphore to prevent too many parallel processes from running This method is deprecated because multiple processes that are completely unrelated could end up using the same semaphore. To prevent this, use `NamedBoundedSemaphore` instead. The `NamedBoundedSemaphore` is a drop-in replacement for this class. >>> semaphore = BoundedSemaphore(2, directory='') >>> str(semaphore.get_filenames()[0]) 'bounded_semaphore.00.lock' >>> str(sorted(semaphore.get_random_filenames())[1]) 'bounded_semaphore.01.lock' Lock | Nonerbounded_semaphore{name}.{number:02d}.lockTc>XlX lX0lX@lSUl[ TU]UUUS9 U(aUS:Xa[R"S[SS9 gg)N)r?r@rBrz_`BoundedSemaphore` without an explicit `name` argument is deprecated, use NamedBoundedSemaphorerr) maximumr2filename_pattern directoryrrrIrrDeprecationWarning rHrr2rrr?r@rBrs rrIBoundedSemaphore.__init__sf  0"!%  )-  t22 MMD"  3rct[UR5Vs/sHoRU5PM sn$s snfr)ranger get_filename)rHns r get_filenamesBoundedSemaphore.get_filenamess..3DLL.AB.A!!!$.ABBBs5cd[UR55n[R"U5 U$r)listrrandomshuffle)rH filenamess rget_random_filenames%BoundedSemaphore.get_random_filenamess'++-. y!rc[R"UR5URR UR US9- $)N)r2number)r%r&rrformatr2)rHrs rrBoundedSemaphore.get_filenames?||DNN+d.C.C.J.J/K/   rc\UR(aS5eUR5nURX5H=n[R SXT5 UR U5(dM1URs $ [ UUR5=n(a[R"5eg)NzAlready lockedztrying lock (attempt %d) %r) rrrXloggerdebugtry_lockrrBrr)rHr?r@rBrrs rrLBoundedSemaphore.acquires 99...}&&( ((AA LL6 E}}Y''yy B (   ! !   **, ,rcUHYn[RSU5 [USS9UlURR 5 [RSU5 g g![ R a SUlM|f=f)Nztrying lock for %rT)rBz locked %rF)rrr rrLrr)rHrr7s rrBoundedSemaphore.try_lockst!H LL-x 8X=DI  !!# [(3" ++ !   !sA!!B?BcbURb"URR5 SUlggr)rr\r[s rr\BoundedSemaphore.releases( 99 II   DI !r)rrrrr2)rintr2r,rr,rr,r?ror@rorBrprqrr)rqztyping.Sequence[pathlib.Path])rrrqz pathlib.Pathrn)r?ror@rorBrprqr)rztyping.Sequence[Filename]rqrAru)rxryrzr{rr|r* gettempdirrDrErIrrrrLrr\rrrs@rrrs   ( :!,,. /'=(,     %& :C  !%'+(, %&   0 rrc|^\rSrSrSrSS\R "5\\S4SU4Sjjjr Sr U=r $) NamedBoundedSemaphoreia] Bounded semaphore to prevent too many parallel processes from running It's also possible to specify a timeout when acquiring the lock to wait for a resource to become available. This is very similar to `threading.BoundedSemaphore` but works across multiple processes and across multiple operating systems. Because this works across multiple processes it's important to give the semaphore a name. This name is used to create the lock files. If you don't specify a name, a random name will be generated. This means that you can't use the same semaphore in multiple processes unless you pass the semaphore object to the other processes. >>> semaphore = NamedBoundedSemaphore(2, name='test') >>> str(semaphore.get_filenames()[0]) '...test.00.lock' >>> semaphore = NamedBoundedSemaphore(2) >>> 'bounded_semaphore' in str(semaphore.get_filenames()[0]) True NrTc l>UcS[R"SS5S3n[TU] UUUUUUU5 g)Nzbounded_semaphore.ri@Bd)rrandintrrIrs rrINamedBoundedSemaphore.__init__6sF <'q'(B1'EFD         rr)rrr2z str | Nonerr,rr,r?ror@rorBrprqrr) rxryrzr{rr*rrDrErIrrrs@rrrs}6 :!,,. /'=(,         % &    rr)rrwrrwrqrw)T)r7r r8rArqztyping.Iterator[types.IO])) __future__rr}rr3loggingr.r%rr*rOrrrrrrr r getLoggerrxrrDrErFr EXCLUSIVErr__all__rcontextmanagerr ABCr<r rrrrrrrrs "    77!   8 $ !!++i.A.A.N.NN    :>F2 5$5$ 5$5$5$pAswwAHn8nb1!D1!h%%2gxgT- ,- r