Stop Mixing FastAPI with Business Logic: Fix It with Ports & Adapters

Here's an example of a fast API app for placing orders. And this is the endpoint. Uh this internally calls a place order function that gets a database connection store keeping a unit and a quantity. Uh and then it does some things like executing a few SQL queries. It does some checks. Uh it checks whether the stock is available and if so then it's going to update the inventory by executing another database uh statement. The problem here is that we have a bunch of business logic here that is directly integrated with both fast API and SQL alchemy because it depends on a database. It raises HTTP exceptions. So in essence, this domain that we're modeling here is not a domain. It's a complete mess. Now today I'll show you a refactor that removes these types of framework dependencies. It's called ports and adapters. and I'm going to do it step by step so you can directly apply it to your own project. Now, if you've run into problems like you see in this code, but you never had a systematic way to fix it, that's exactly why I'm working on a brand new program called Software Design Mastery. This is not just another online course. It goes very deep and it's way more than just videos. If you want to be the first to know when it opens, trust me, you don't want to miss this. Join the wait list for free at iron. co/mastery. The link is also in the description.

Now, what is the actual problem in this code? We're essentially mixing domain logic with API OM framework code and that creates a lot of complexity and concrete symptoms of that are that within this place order function we are raising HTTP exceptions which is a uh transport concern. We have uh the function that accepts a database connection. So the database in this case becomes like a mandatory thing even if you want to run a unit test or you have to do like complicated mocking and patching which is really annoying. Also the return type here if you take a look it returns a dictionary but that's structured exactly like what the API is going to send back uh as a result to the user. So the sort of domain language that you might want to have it disappears because we're directly coupling it to the API. And that means that if you now want to change frameworks, let's say you want to switch to a NoSQL database or maybe you want to move away from fast API to something else, um you have to refactor completely this code. You have to rewrite basically everything. Now you might be thinking who cares, right? I can just ask my claw code agent to simply rewrite all of that. I don't need to worry about it. But the problem with that line of thinking is that even AI coding agents have some sort of context. And the more you do these types of things, the more complex, the bigger that context becomes and the worse the output of the AI agent is going to be. So even if you're not writing the code directly yourself, but you let an AI do that job for you, you still need to think about decoupling things and applying the right architectures to make sure that you don't mix up all the frameworks and the business logic and that's what we're going to address in today's video. Now

ports and adapters, I mentioned this architecture. It basically means that you split code into three layers with a pretty strict direction of dependencies. One is you have the domain. So this is kind of pure isolated thing where you have rules decisions invariance. You don't want to import things from fast API, SQL alchemy, pyantic or other types of frameworks that your the rest of your application depends on. Then you have ports. These are basically the interfaces the definition of the interfaces that your domain is going to need. In Python you can do that with ABCs or with protocols. And the domain defines these things. For example, you may have an inventory port which defines how you interact with an inventory or you may have a payment port or an email port. And finally, you have adapters. So these are specific implementations for particular frameworks. Uh in this case, we're dealing with stock and inventory. So you might have a SQL alchemy adapter that implements the inventory port. And in a sense, you'll have another adapter which is fast API that translates uh HTTP things into domain types or errors.

So the demo use case that I'll do today is placing an order. And I already showed you part of the code. So this is the actual API. There's a router. Uh it has an input for placing the order. So it needs a SQU and a quantity. And there's also an output which gives you an SQU, quantity, and remaining stock. We have the place order function that currently gets a database connection and other things that it needs. Then we have the business logic. Then I have the endpoint that's part of the router. I also have a database file. So this is sort of a simple wrapper around SQL alchemy. There's a get DB which is used as dependency in fast API. And I have an init DB which creates the database and it also inserts a couple of items into the database just for testing. And finally, I have the main file where everything is patched up. So when I start the app, then this is what you see. And what I can do now is post a request to the API. For example, here I'm placing an order like so. And then you see we get a quantity of three. The remaining stock is seven. Initially it was 10. So uh this seems to work as we expect. And by the way, this is a cool little trick if you want to be able to read JSON response from your curl request very easily if you have Python installed, which I assume you have if you're watching this video. The first

thing we need to do is to decouple the domain from all of the frameworks. And we can do that by basically giving it its own language and specifying the types and the errors that we need to deal with in the domain. So let's take a look at what we actually need now in the API file. So this is currently where the business logic is. You see that we have a couple of cases where we raise exceptions. For example, uh we need a positive quantity or uh SQU is unknown. This is another possibility. This also an error that can happen in the domain or third type of error is that actually a particular SKU is out of stock. So these are the types of errors that we need to deal with. So as a first step, let's create a folder domain where we're going to put all the domain specific things. And inside that domain, I'm also going to add a dunder init file like so. So the first thing that we can do is define the errors that we have in the domain like so. The first thing that I'm going to do is create a class domain error which is a subclass of exception. So this is going to be the superass of all my specific domain errors that I have to deal with. For example, we will have an invalid quantity. I'll leave these classes empty for the time being. We also have an unnown SKU which is another domain error. And by the way, you can store extra information in these errors. For example, if I create an initializer like so, I can pass the SQU that is unknown. And then I can initialize the superass by passing a message like so. And then we can store that in the object so you can refer to it later on. Or here's another example out of stock. It's also a domain error. That was the third type that we had to deal with. And there I can do something very similar in that we store the SKU requested and available. And I provide a message that's passed to the initializer of the superass. So these are the errors that we can have in our domain. Maybe later on we can add more, but at least this is a good basis. Now next to the errors, we're also going to have models. These are in essence the objects that are part of our domain. And let me use a data class for that. As you can see on now, I'm not importing any SQL alchemy or fast API stuff. I don't really care about that, right? We're dealing with purely the domain. That's the whole idea. So, I'm going to have a class order request. And let's turn this into a data class. We can even make it frozen. order request and let's say that's going to have a SQU. You can also define uh alias types here if you want to. And we have a quantity which is an integer. We can add another data class called order placed. Let's also make that frozen. And this is also going to have an SKU that's going to be a quantity and a remaining stock like so. So these are the data types that we need to deal with in the domain and these are all independent of frameworks. Right? If for example I swap a fast API for something else then these types are not going to change. If I swap the database for something else they're still going to be the same types. And

what you can do now in the API is now take this place order function and start moving out these HTTP exceptions because we've now defined the errors. So from domain dot errors I'm going to import let's say the out of stock error the invalid quantity and unknown SQ errors like so. And now what we can do is instead of raising this HTTP exception, we're going to raise a quantity error. So here an invalid quantity error. And then this one we need to move out of the place order function. And where do we put it? Well, actually what we will do is in our endpoint we put this in a try except block like so. And there we can check for an invalid quantity error. like so and then we can raise an HTTP exception and there we pass the error as a string like so. So now we have introduced this translation layer. We're actually placing an order that contains the business logic. It doesn't raise an HTTP exception but uh we handle that in our endpoint where we translate these domain exceptions into actual HTTP exceptions. So this is where we introduce basically the framework and we can do the same thing for the other parts of place order. So here for example we have an unknown SKU. So I can raise an unknown SKU error and I need to reimpport that because I have automatic cleanup of imports which is not always very helpful. So unknown SKU that gets the SQU that we ask for. So I simply pass that as an argument here and I remove this HTTP exception and then I put it here except the unknown SQ and then I'm simply going to raise that right here like so. And finally we have the third type of error and that's this one. So here we're going to raise an out of the stock exception. And this is going to need the SQ. It's requested quantity and the available quantity like so. And then this part we can put in another except block. like so. And actually this return statement we can put it also below the try accept block like so. So now here all the translation from the domain errors to the HTTP errors take place and in our domain function place order we don't have to deal with that anymore which is exactly what we want. Now I also defined the data types for the order. So I'll get back to that in a

minute. But the first thing that we need to do is to define a port which is the next ingredient of the ports and adapters architecture. So I have to specify what a domain actually needs and in fact what it needs is some way some interface to manage the dependencies. Right? So what I'm going to do is in my domain folder I'm going to create a ports file. And here I'm going to write the protocol for an inventory port. So from typing I'm going to import the protocol because we will need that. And then I'm going to create a class inventory ports which is a protocol like so. And if you take a look at the uh logic in place order. So there's actually two things that happen here. The first is that we check whether stock is available. So that's this first part. And the other thing is that we actually update the inventory and set the new stock. So these two things we need to do in the port. So let's say we have a get stock method. We pass a store keeping unit and well this is going to return an integer like so. That's the first. The second one is called let's say reserve. This gets an SPU. it gets a quantity and it's going to return the remaining stock as a result of the reservation. So that's our port. That's the only thing that actually the domain needs and the direction of dependencies is important here. So the domain itself depends on abstractions namely an inventory port and the infrastructure that's all the frameworks and everything they will depend on the domain. And now what's

nice is we can take basically this whole uh place order function and we can rewrite it as pure domain logic because all of these uh specific things that we have here we don't need to do them uh as part of the business logic because we have the port. So I'll leave this function here for the time being. But what I'll do is I'll copy over all of this code and then I'm going to create in my domain folder a use cases file that contains the actual use case. I'll just copy this here for the time being. Now of course we don't have all of these dependencies here. So let's clean that up. So first from errors we're going to import in quantity unsq and out of stock. So that already solves quite a few of these uh import issues. The next thing that we're going to do is that we will fix the inputs that this gets. So the first thing is that this uh gets an SKU and a quantity. So in fact we defined an order request that represents that input. So from models we're going to import order request and we will also need order placed which is going to be the return value of this use case. So this is not simply going to return some kind of dictionary. It will return an order placed object and as an input this is going to get a order request like so. So this we don't need but then next thing in order to place the order we need the interface which is the port that we just defined. So from port we're going to import the inventory port and our place order gets the inventory like so. There we go. And now we can in essence rewrite all this logic so that purely relies on what we get here as an argument. So first check we need to check that the quantity is valid. So there is this and then here this is actually the selection. So what we'll do here is we get the available stock which is going to be inventory dot get stock for the SKU and then we can check that if available is less than the quantity then we can raise the out of stock like we have here. And that's going to get the SKU. It's quantity. And I see we actually forgot to add one more thing to our inventory port, which to check whether an SKU exists. So we will add it right here. And this is going to return a boolean like so. And now back to our use case. So before I check this availability, what I can do is if uh not inventory. exist SKU and that needs to be the SKU from the request, then we're going to raise the unknown SQU error like so. This is what we can remove. I can also remove this. And then finally we need to do this which is to update the stock. So then what we're going to do is remaining equals inventory dot reserve skew and the quantity like so. And then finally we can return auto placed like so and we don't need any of this code anymore. So this is really nice because now we have modeled the business logic based on the types that are available in the domain and the inventory port. We have zero knowledge of frameworks here. There's no SQL alchemy. There is no fast API. There's nothing. We purely have the business logic here model inside the domain. Also, if you look at the imports, you see that these are purely imports from the domain. The domain is really isolated right now. And this is really nice because now we can write tests for this by passing a mock uh inventory port for example or a sample order request. We don't have to deal with the database connection. It's all very simple. Now the final thing we need

to do is actually build an adapter for SQL Alchemy so that we can actually pass it in as a port to our logic that we have right here. So for that I'm going to create a new folder and let's call that adapters. And that doesn't need to be part of the domain. That should actually top level. And within that folder I will also add an initializer. And I'm going to add a SQL alchemy inventory adapter like so. And this is where we're going to put the actual SQL alchemy code. So I'll use again a data class for that because that's simple and easy to set up. We're going to have a class SQL alchemy inventory adapter which is an inventory port. Let's import that. And by the way, you don't have to do that because it's a protocol. But in this way, you explicitly say that there is an inheritance relationship. And this should also be a data class. So this is going to have the actual uh database connection. So from SQL alchemy dot engine we're going to import the connection type. And then we have to implement the interface. So there is a uh get stock method and I've of course already implemented that in the original version. We're still missing a couple of imports here. Here we have the method for checking whether an SKU exists. And finally we're going to have the reserve method. And this is in essence the same as we had in the logic before. So this contains now just the database stuff. And now the only thing we need to do is patch things up in the API. So have my API here. So instead of this monstrosity, I'm going to delete that immediately. What we're going to do is from the uh domain use cases we're going to import place order and from the models we're going to import order request and order placed because we will need those two things to communicate with our domain. So that's for the domain and then the final thing we will need is from adaptersql alchemy we need to import the SQL alchemy inventory adapter and then what we'll do in our endpoint is basically patch all of these things off. So we have the adapter which is a SQL alchemy inventory adapter and that gets the connection like so and then place order. It's going to need the adapter. I'll remove this and order request object and this gets a model dump of the payload like so. So you see we patch up everything. This is where we have the specific adapters and how it's connected with the domain. But the domain itself is completely independent. The SQL alchemy inventory adapter is also independent from the domain logic. It simply knows how to deal with the database. So it's all very nicely separated. Final thing we need to do is of course here place order out. So we need to translate the result to this object and that is actually also very easy. So we can simply say that SKU is the result dot SKU the quantity equals the result dot quantity and finally remaining stock equals the remaining stock from the result like so. And that is all there is to it. So now let's restart the server. I've also deleted the database so it's reinitialized. And now I can send another curl request. So as you can see we have remaining stock seven. Let me try that one more time. And now we have remaining stock four. If I now try to reserve a quantity of let's say 30 items, we're going to get this error. Or let's put this back to three. If I change this to some other kind of SKU, then we get the unknown SKU error. Or let's say quantity is minus 12. Actually, this is caught at the API level by the pyantic model. So, even though I put this check in the domain, it's also handled on the API level. You could opt to remove it there. So, that's ports and examples. Now, in production, you'll probably do uh things a bit differently. Maybe you want to make reserving stock atomic, but that is actually an improvement in the adapter. The domain basically stays untouched.

And if you look at the endpoint, it does now exactly three things. It parses the inputs. That's pedantic. It calls the domain use case with the port and it maps the domain errors to HTTP responses. Honestly, I really like this

pattern for APIs. It makes for a very clean solution and also the business logic is now way easier to test. You can write a fake inventory adapter, for example, then test the business logic with that. It's just really cool. But I'd like to hear from you. Have you tried ports and adapters before? What were your experiences with it? Do you have any tips that you would like to share? Let me know your thoughts in the comments. Now, if you enjoyed this video, you might also like this video I did recently covering CQS, another great pattern for backend APIs. Thanks for watching and see you next