Python/Flask best practice for service layer

Question

I'm a little new to Python ( I come from a Java/C++ background ). I've been playing around with Flask for web development. My question relates somewhat to dependecy injection and thread safety. In the Java/Spring word you would have a Controller that has a Service say UserService injected into it. And when you have say a addUser endpoint it would call the userService.addUser(someData).

If I want to do the same in Python/Flask is it best practice to just have a file called UserService with functions like addUser(), deleteUser() etc and just straight call them with UserService.addUser(), UserService.deleteUser() and is this thread safe? Or should I have a new instance of User service in each endpoint?

Answer 1

As with all things thread-related, the question is "did you make it thread-safe"?

If your user service looks like this:

# user_service.py
from some.package import database

def add_user(user_information=None):
    db = database.connect()
    db.insert(user_information)

def update_user(user_information=None):
    db = database.connect()
    db.update(user_information["user_id"], user_information)

def delete_user(user_id=None):
    db = database.connect()
    db.delete(user_id)

Then, assuming an even remotely sane implementation of some.package.database it will be thread safe. If, on the other hand, you do something like this:

# bad_user_service.py
from some.package import database

# Shared single connection
# Probably *not* thread safe
db = database.connect()

def add_user(user_information=None):
    db.insert(user_information)

# ... etc. ...

Now, including db = database.connect() at the top of each of your service methods is very un-DRY. You can avoid that issue by wrapping up your connection-specific work in a decorator (for example):

def provide_db(func):

    @functools.wraps(func)
    def new_function(*args, **kwargs):
        db = database.connect()
        return func(db, *args, **kwargs)

    return new_function

Then you could do this:

# user_service.py
from your.data.layer import provide_db

@provide_db
def add_user(db, user_information=None):
    db.insert(user_information)

# ... etc. ...