Allows existing functions to be run and stopped in seperate processes
Project description
SingletonProcess
A SingletonProcess is a function that is run in a seperate process and only one exists at a time. This is useful for use facing functions that must run in the background and need to be shutdown automatically.
Always guard your code with if __name__ == '__main__'
or multiprocessing will fail.
Note: This module handles one specific use case very well. In high performance applications or different use cases, you may be better off with a custom solution.
Examples
See examples for more examples.
Run two processes simultaneously (notice the unique process ID)
from SingletonProcess import SingletonProcess, block
from time import sleep
@SingletonProcess
def longComputation(x):
sleep(5)
return x ** 2
if __name__ == '__main__':
a = longComputation(1, pid='a')
b = longComputation(2, pid='b')
block()
print(a, b)
Stop new process automatically (notice pid=None, which acts as a wildcard and stops all processes)
from SingletonProcess import SingletonProcess, block
@SingletonProcess
def uniqueProcess():
while True:
print("Doing important stuff!")
if __name__ == '__main__':
uniqueProcess()
uniqueProcess() #stops execution of first process
block()
Use VBSingletonProcess
and verbose = True
to see
detailed info about internal proccesses.
from SingletonProcess import VBSingletonProcess, block, terminateProcessesByPID
from time import sleep
@VBSingletonProcess
def printList(l: list):
for item in l:
print(item)
sleep(1)
if __name__ == "__main__":
printList([1, 2, 3, 4, 5, 6], pid='a')
printList([10, 20, 30, 40, 50, 60], pid='b')
sleep(2.5)
printList(['a', 'b', 'c', 'd'])
sleep(2.5)
printList(['hello', 'hello', 'world', 'world'], pid='c')
sleep(2.5)
printList(['so', 'long', 'and', 'thanks', 'for', 'all', 'the', 'fish'], pid='c')
printList(range(0,100), pid='d')
block(pid='c', verbose=True)
terminateProcessesByPID(pid='d', verbose=True)
block(verbose=True)
Use different poolgroups to seperate out different types of tasks.
from SingletonProcess import SingletonProcess, block
class GroupA(SingletonProcess):
poolgroup = 'A'
class GroupB(SingletonProcess):
poolgroup = 'B'
@GroupA
def uniqueProcessA():
while True:
print("Doing important stuff!")
@GroupB
def uniqueProcessB():
while True:
print("Doing other important stuff!")
if __name__ == '__main__':
uniqueProcessA()
uniqueProcessB() #first process still runs
block(poolgroup='A')
block(poolgroup='B')
You can also override the getPID
class method for custom use cases.
from SingletonProcess import SingletonProcess, block
from time import sleep
class SafeServer(SingletonProcess):
@staticmethod
def getPID(args, kwargs):
return kwargs['hostname']
@SafeServer
def startServer(hostname):
print("Starting server on hostname: ", hostname)
while True:
pass
if __name__ == '__main__':
startServer(hostname='https://examplea.com')
startServer(hostname='https://exampleb.com')
sleep(1)
startServer(hostname='https://examplea.com') #stops first server
block()
ThreadSafeSingletonProcess
Uses spawn instead of fork on linux, which may work better for projects that also use threads.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for SingletonProcess-0.1.0rc8.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 17d5e4a29cf3f3973ba6236fdc6a12b702b2a4e203924d9bee8c16a7dbeed302 |
|
MD5 | 5b6bc790ec0e767f1539b6fdd9f7b043 |
|
BLAKE2b-256 | bf2dff296aa29ee6d4fd91f2efc9b1510ccac97efac9d81fad775c1b57e68078 |
Hashes for SingletonProcess-0.1.0rc8-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | a622de7735130c527fc1c4e7268332df5e868b8be5d0a43e875e2618c785779c |
|
MD5 | 96f06de356c76b9b3e9173bf088ad6a9 |
|
BLAKE2b-256 | 5b6ec8ccb3bd295fa2767f6ca7b2f3cc3017858d41235d63a8febb83fc4d5539 |