Background Function in Python

background function in Python

Do something like this:

def function_that_downloads(my_args):
    # do some long download here

then inline, do something like this:

import threading
def my_inline_function(some_args):
    # do some stuff
    download_thread = threading.Thread(target=function_that_downloads, name="Downloader", args=some_args)
    download_thread.start()
    # continue doing stuff

You may want to check if the thread has finished before going on to other things by calling download_thread.isAlive()

Python functions run as background processes

This sounds like threading.

The threading module is relatively simple, but beware doing any CPU bound activities that way. With lots of timers and sleeps, or IO bound then this is fine.

Threading module example:

import time
from threading import Thread

def timer(name):
    count = 0
    while True:
        time.sleep(60)
        count += 1            
        print("Hi " + name + "This program has now been running for " + str(count) + " minutes.")

print("Hello! Welcome to the program timer!")
name = raw_input("What is your name?")
print("Nice to meet you, " + name + "!")
background_thread = Thread(target=timer, args=(name,))
background_thread.start()

To Clarify - relatively simple compared with the multiprocessing module - which you may want to look at if your background activity is CPU heavy.

The python docs have a good tutorial on this at: https://docs.python.org/2/library/threading.html#thread-objects

how to run background function after response return from web.py

Have a look python documentation for Thread.run()

Note:
With background task you won't be able to return "Exception occurred" as you're doing now. I believe you're OK with it.

Here's a small easy solution. There are other ways too but I feel you should explore more by yourself since you're a python beginner. :)

import web
import time 

urls = (
    '/', 'index'
)

app = web.application(urls, globals())

class index:

    def writeToFile():
     try:
       with open('filename.txt', 'a') as file:
            for i in range(100):
                time.sleep(1)
                file.write("No of times: {}".format(i))
       # Log completion
    except:
        # Log error

    def GET(self):
        thread = Thread(target=writeToFile)
        thread.start()

        return {<myJSON>}

if __name__ == "__main__":     
     app.run()

How to run a function in 'background'

In asyncio you might use two coroutines, one that reads from file, and the other that processes the file. Since they communicate using queue, they don't need the global variable. For example:

import os, asyncio

async def tail(f, queue):
    f.seek(0, os.SEEK_END)
    while True:
        line = f.readline()
        if not line:
            await asyncio.sleep(0.1)
            continue
        await queue.put(line)

async def consume(queue):
    lines = []
    while True:
        next_line = await queue.get()
        lines.append(next_line)
        # it is not clear if you want somefun to receive the next
        # line or *all* lines, but it's easy to do either
        somefun(next_line)

def somefun(line):
    # do something with line
    print(f'line: {line!r}')

async def main():
    queue = asyncio.Queue()
    with open('file.txt') as f:
        await asyncio.gather(tail(f, queue), consume(queue))

if __name__ == '__main__':
    asyncio.run(main())
    # or, on Python older than 3.7:
    #asyncio.get_event_loop().run_until_complete(main())

The beauty of an asyncio-based solution is that you can easily start an arbitrary number of such coroutines in parallel (e.g. you could start gather(main1(), main2()) in an outer coroutine, and run that), and have them all share the same thread.

How to run function in background in Django view

Maybe you can take a look at the async support. async

Is there a way to run a function in the background of a python flask app?

Thank you for your suggestions, as they definitely helped me research the topic, but I finally figured out how to get threading working.

Turns out my issue was returning the value to a list for the main program to use.

By providing the list notes_pressed as an argument to get_next_note(), it ensures that the list is updated when return_pressed_notes() is called.

Here is the code I used to solve my issue:

from Flask import Flask
import mido
from threading import Thread

app = Flask(__name__)

# List to store pressed keys
notes_pressed = []

@app.route('/get_notes', methods=['GET'])
def return_pressed_notes():
    return json.dumps(notes_pressed)

# Function to translate midi key numbers to note letters
def translate_key(key_num):
    ...

# Function that returns recently played note
def get_next_note(notes_pressed):
    # Open port to listen for note presses
    with mido.open_input() as inport:
        # Retreive key presses from port
        for msg in inport:
            # If key press is valid
            # - note is pressed
            # - velocity!=0 (prevents ghost notes)
            if (msg.type=='note_on' and msg.velocity!=0 and msg.channel==0):
                # Add new note to list
                notes_pressed.append(translate_key(msg.note - lowest_key))

# Run main program
if __name__ == '__main__':
    # NEW CODE
    p = Thread(target=get_next_note, args=(notes_pressed,))
    p.start()
    app.run(debug=True, use_reloader=False)
    p.join()

FastAPI python: How to run a thread in the background?

You should start your Thread before calling uvicorn.run, as uvicorn.run is blocking the thread.

PS: In your question you state that you would like the background task to run every 5 minutes, but in your code you say every 5 seconds. The below examples assume that is the latter you want. If you want it to be executed every 5 minutes instead, then adjust the time to 60 * 5.

Option 1

import time
import threading
from fastapi import FastAPI
import uvicorn

app = FastAPI()
class BackgroundTasks(threading.Thread):
    def run(self,*args,**kwargs):
        while True:
            print('Hello')
            time.sleep(5)
  
if __name__ == '__main__':
    t = BackgroundTasks()
    t.start()
    uvicorn.run(app, host="0.0.0.0", port=8000)

You could also start your thread using FastAPI's startup event, as long as it is ok to run before the application starts.

@app.on_event("startup")
async def startup_event():
    t = BackgroundTasks()
    t.start()

Option 2

You could instead use a repeating Event scheduler for the background task, as below:

import sched, time
from threading import Thread
from fastapi import FastAPI
import uvicorn

app = FastAPI()
s = sched.scheduler(time.time, time.sleep)

def print_event(sc): 
    print("Hello")
    sc.enter(5, 1, print_event, (sc,))

def start_scheduler():
    s.enter(5, 1, print_event, (s,))
    s.run()

@app.on_event("startup")
async def startup_event():
    thread = Thread(target = start_scheduler)
    thread.start()

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8000)

---

Related Topics