Django Expense Manager 8 - Implementing the API for Transactions
I’m back, I think. What are we going to do today? We will implement a new endpoint to be able to read and add Transactions. You might wonder why we didn’t do it all in the previous post. I wonder the same thing. Actually, the reason we didn’t is because in the previous post I did not want to touch upon hyperlinks between our entities. In this one, I do. So what do I mean when I say we are going to hyperlink our API?
It refers to dealing with the relationships of our entities: transactions, categories, retailers (for now). For example, if I call an API endpoint that returns a specific transaction, I might want to include links in there to the API endpoints that would return the category details or the retailer details. For example, let’s say we call the (not yet created) transactions API endpoint:
curl 'http://localhost:8000/api/transactions/' -H 'Authorization: Bearer SUPERSECRETAMAZINGTOKEN'
the response could beL
[
{
name: "SampleTransaction",
amount: "200.00",
amount_currency: "MDL",
retailer: 1,
category: 1,
date: "2022-02-01",
transaction_type: "E",
recurring: false,
user: 2,
},
];
We get that the retailer is 1 and category is 1. Well, that’s not very useful is it? In order to know what the category is, we then need to call the (not yet created) detail retailer API endpoint and get the name from there. Is there a better way (or more descriptive way)? You bet! For that, we would need to use some of the other possible relations between the models. We are going to use the StringRelatedField to change that so it says:
[
{
name: "SampleTransaction",
amount: "200.00",
amount_currency: "MDL",
retailer: "Abbre",
category: "Food",
date: "2022-02-01",
transaction_type: "E",
recurring: false,
user: 2,
},
];
The target of the relatioship (i.e. the Retailer or Category) will be rerpesented using its __str__ method so let’s make sure both models have something we are happy with. Since we’re here, let’s also give it a nice __repr__ for when we’re working in the shell. In expenses/models.py,
...
class Retailer(models.Model):
...
def __str__(self):
return f"{self.name} ({'Online' if self.online else 'Physical'})"
def __repr__(self):
return f"<Retailer: {self.name}>"
class Category(models.Model):
....
def __str__(self):
return f"{self.name} ({self.get_product_type_display()})"
def __repr__(self):
return f"<Category: {self.name}>"
...
Now, let’s look at the actual Transaction serializer (expenses/api/serializers.py):
class TransactionSerializer(serializers.ModelSerializer):
name = serializers.CharField(max_length=255)
amount = MoneyField(max_digits=14, decimal_places=2)
amount_currency = serializers.ChoiceField(choices=CURRENCIES)
retailer = serializers.StringRelatedField(
read_only=True,
)
category = serializers.StringRelatedField(
read_only=True,
)
date = serializers.DateField()
transaction_type = serializers.ChoiceField(choices=Transaction.TRANSACTION_TYPES)
recurring = serializers.BooleanField()
user = serializers.PrimaryKeyRelatedField(
read_only=True, default=serializers.CurrentUserDefault()
)
class Meta:
model = Transaction
fields = ["name", "amount", "amount_currency", "retailer", "category", "date", "transaction_type", "recurring", "user"]
What are the important things to consider? Well, the amount and amount_currency are separate fields, with amount being imported (from djmoney.contrib.django_rest_framework import MoneyField) read more here. In the actual Transaction model, the amount is a MoneyField. In the serializer, the two fields are treated separately so we need both of them. We also made amount_currency a ChoiceField with the choices coming from the library upon which djmoney is built.
Let’s also look at the TransactionList view (also inheriting from the ListCreateAPIView generic class).
class TransactionList(generics.ListCreateAPIView):
authentication_classes = [JWTCookieAuthentication]
permission_classes = [IsAuthenticated]
serializer_class = TransactionSerializer
def get_queryset(self):
"""
This ensures only the retailers created by the user are returned.
"""
user = self.request.user
return Transaction.objects.filter(user=user)
def perform_create(self, serializer):
serializer.save(user=self.request.user)
Hold on, we are creating a Transaction object or serialising one but…how does it know to combine the amount and amount_currency fields? Before we look at that, let’s quickly look at serialising/deserialising a model field. The .to_representation() method is used to serialise the data whereas the .to_internal_value() is used to deserialise the data. Ok back to the MoneyField. If we look at this extract from its source code:
...
def to_representation(self, obj):
"""
When ``field_currency`` is not in ``self.validated_data`` then ``obj`` is an instance of ``Decimal``, otherwise
it is ``Money``.
"""
if isinstance(obj, MONEY_CLASSES):
obj = obj.amount
return super().to_representation(obj)
def to_internal_value(self, data):
if isinstance(data, MONEY_CLASSES + (_PrimitiveMoney,)):
amount = super().to_internal_value(data.amount)
try:
return Money(amount, data.currency)
except CurrencyDoesNotExist:
self.fail("invalid_currency", currency=data.currency)
return super().to_internal_value(data)
def get_value(self, data):
amount = super().get_value(data)
currency = data.get(get_currency_field_name(self.field_name), self.default_currency)
if currency and amount is not None and not isinstance(amount, MONEY_CLASSES) and amount is not empty:
return _PrimitiveMoney(amount=amount, currency=currency)
return amount
...
we can see that when it is deserialised, if it’s already an instance of these MoneyClasses, it deserialises the amount. Otherwise, it calls the super, DecimalField, to_internal_value with data. Why did I also include the get_value method in the code snippet above then? DRF documentation says:
Now if you need to access the entire object you’ll instead need to override one or both of the following: Use get_attribute to modify the attribute value passed to to_representation(). Use get_value to modify the data value passed to_internal_value
If we look at the Serializer class in the DRF source code, we see that in the to_internal_value method calls:
for field in fields:
validate_method = getattr(self, 'validate_' + field.field_name, None)
primitive_value = field.get_value(data)
Ok so we access the field’s get_value method, where the currency is extracted as:
currency = data.get(get_currency_field_name(self.field_name), self.default_currency)
where the get_currency_field_name does the following:
def get_currency_field_name(name, field=None):
if field and getattr(field, "currency_field_name", None):
return field.currency_field_name
return "%s_currency" % name
Remember that self.field_name refers to the field_name of “amount”. We are not gonna look into the first two lines because they are called when using the validator. What we care about is the last line -> "%s_currency" % name". εὕρηκα!!!
Ok aside over, that’s how it knows how to deserialize the MoneyField. I honestly forgot what I was even writing about before this. Oh right, we were talking about the Transaction Serializer. There’s an issue there. In fact, there are a few issues. Can you spot what they might be? Let’s first look at the category field. We have it as:
category = serializers.StringRelatedField(
read_only=True,
)
First of all, it’s only read_only. It means that when we POST to this endpoint to create a transaction, we would not be able to specify the category, we don’t want that. That removes the StringRelatedField as a viable option because it’s read_only. Instead, we will use the SlugRelatedField. It does, however, say that we want to ensure that the slug field corresponds to a model field with unique=True. We don’t currently have that in our models. Looking at our models, it wouldn’t make sense for any of name, online, user to be unique for the Retailer (for example). There might be multiple Retailers with the same name but from different users. Ok so what we really want is to limit the QuerySet (of the serializer) to just the Retailers of a specific user (still, that doesn’t really prevent there being multiple Retailers with the same name) AND create some kind of Unique Constraint saying there can only be one Retailer with the same name for a specific user (and the same for Category). Let’s first do the Django related changes first and then the DRF related changes.
In our Retailer and Category models, let’s add some UniqueConstraint. At the same time, let’s add some slugs as well so we can use that in serializing maybe. In expenses/models.py:
class Retailer(models.Model):
...
slug = models.CharField(max_length=255)
class Meta:
...
constraints = [
models.UniqueConstraint(fields=["name", "user"], name="retailer_unique_name_per_user"),
models.UniqueConstraint(fields=["slug", "user"], name="retailer_unique_slug_per_user"),
]
class Category(models.Model):
...
slug = models.CharField(max_length=255)
class Meta:
...
constraints = [
models.UniqueConstraint(fields=["name", "user"], name="category_unique_name_per_user"),
models.UniqueConstraint(fields=["slug", "user"], name="category_unique_slug_per_user"),
]
Given that we’ve changed our models, we need to create some migrations.
./manage.py makemigrations --name=adding_slugs_and_unique_constraints expenses
That will create a migration in our expenses/migrations folder. Now, let’s change/add some tests. First, let’s change our RetailerFactory and CategoryFactory to add a slug field:
class RetailerFactory(factory.django.DjangoModelFactory):
class Meta:
model = Retailer
user = factory.SubFactory(UserFactory)
name = factory.Sequence(lambda n: f"AmazI{n}")
slug = factory.Sequence(lambda n: f"retailer-slug-{n}")
online = True
class CategoryFactory(factory.django.DjangoModelFactory):
class Meta:
model = Category
user = factory.SubFactory(UserFactory)
name = factory.Sequence(lambda n: f"Category_{n}")
slug = factory.Sequence(lambda n: f"category-slug-{n}")
product_type = "P"
Then, let’s test that one user cannot create two categories with the same slug nor can they create two retailers with the same slug.
@pytest.mark.django_db
def test_cannot_create_two_categories_with_same_slug_one_user(category_factory, user_factory):
user = user_factory()
_ = category_factory(slug="hi-there", user=user)
with pytest.raises(IntegrityError) as exception_info:
category_factory(slug="hi-there", user=user)
assert exception_info.type == IntegrityError
@pytest.mark.django_db
def test_cannot_create_two_retailers_with_same_slug_one_user(retailer_factory, user_factory):
user = user_factory()
_ = retailer_factory(slug="hi-there", user=user)
with pytest.raises(IntegrityError) as exception_info:
retailer_factory(slug="hi-there", user=user)
assert exception_info.type == IntegrityError
So what was the point of doing all of that? Well remember we said we will use the SlugRelatedField. If we use the this as a read-write field, we need to make sure that the slugfield corresponds to a model field with unique=True. At the same time, remember that our unique is not globally true. Instead, it is unique to a certain user. Consequently, we need to have some kind of way to pass in the user as well.
We will do that by subclassing the SlugRelatedField in order to create a custom relational field. In order to provide a dynamic queryset that changes based on the context, we overwrite the get_queryset method.
class UserSpecificSlugRelatedField(serializers.SlugRelatedField):
def get_queryset(self):
request = self.context.get("request")
queryset = super(UserSpecificSlugRelatedField, self).get_queryset()
if queryset is None:
return None
if request and not request.user.is_superuser:
queryset = queryset.filter(user=request.user)
return queryset
What do we do in there? Well, we start with the base queryset. Then, since we have access to the request from context, let’s get the user from there. We then filter the queryset to only return the user’s object. Now, when a .get is done for the slug, there should be a match. Let’s have a look at our updated TransactionSerializer using our custom field.
class TransactionSerializer(serializers.ModelSerializer):
name = serializers.CharField(max_length=255)
amount = MoneyField(max_digits=14, decimal_places=2)
amount_currency = serializers.ChoiceField(choices=CURRENCIES)
retailer = UserSpecificSlugRelatedField(many=False, read_only=False, slug_field='slug', queryset=Retailer.objects)
category = serializers.StringRelatedField(
read_only=True,
)
category = UserSpecificSlugRelatedField(many=False, read_only=False, slug_field='slug', queryset=Category.objects)
date = serializers.DateField()
transaction_type = serializers.ChoiceField(choices=Transaction.TRANSACTION_TYPES)
recurring = serializers.BooleanField()
user = serializers.PrimaryKeyRelatedField(
read_only=True, default=serializers.CurrentUserDefault()
)
class Meta:
model = Transaction
fields = ["name", "amount", "amount_currency", "retailer", "category", "date", "transaction_type", "recurring", "user"]
Now if we post something like:
curl --location --request POST 'http://localhost:8000/api/transactions/' \
--header 'Authorization: Bearer SUPERSAFETOEN' \
--header 'Content-Type: application/json' \
--data-raw '{
"name": "Lalaa",
"amount": "120.22",
"amount_currency": "XYZ",
"retailer": "Bamzon",
"category": "Subscriptions",
"date": "2022-04-14",
"transaction_type": "E",
"recurring": false,
"user": 1
}'
And we post it again, and again, we create another one each time. Uh, perhaps we’d only like one such unique combination. Let’s add one.
We are going to be using DRF Validators, specifically the UniqueTogetherValidator.
In the TransactionSerializer meta, we add:
class Meta:
model = Transaction
fields = ["name", "amount", "amount_currency", "retailer", "category", "date", "transaction_type", "recurring", "user"]
validators = [
UniqueTogetherValidator(
queryset=Transaction.objects.all(),
fields=fields
)
]
Now, if we post the same thing twice, we get:
{
"non_field_errors": [
"The fields name, amount, amount_currency, retailer, category, date, transaction_type, recurring, user must make a unique set."
]
}
I have also added the same thing for the other two Serializers (well, with the queryset changed of course).
And now it’s time for that? It’s time for tests of course. Let’s start with adding some tests to the Categories and Retailers endpoint, and then we’ll add some Transaction ones.
In tests/api/test_categories.py:
@pytest.mark.django_db
def test_post_category_twice(api_client, user_factory):
initial_count: int = Category.objects.all().count()
user = user_factory()
api_client.force_authenticate(user=user)
list_category_url: str = reverse("view_and_create_categories")
data = {"name": "Entertainment", "product_type": "E"}
list_response: Response = api_client.post(
list_category_url, data=data, format="json"
)
final_count: int = Category.objects.all().count()
assert list_response.status_code == status.HTTP_201_CREATED
assert final_count == initial_count + 1
# now let's try post it again
second_list_response: Response = api_client.post(
list_category_url, data=data, format="json"
)
assert second_list_response.status_code == status.HTTP_400_BAD_REQUEST
json_response: dict[str, list[str]] = second_list_response.json()
assert "The fields name, product_type, user must make a unique set." in json_response["non_field_errors"]
# let's also assert that the count hasn't changed
assert Category.objects.count() == final_count
In tests/api/test_retailers.py, I included a similar test. Now let’s look at the Transaction tests. The format of the tests is similar to retailer/category ones.
We start with checking that authentication is required to access the endpoint:
@pytest.mark.django_db
def test_authentication_required(api_client: APIClient):
list_transaction_url: str = reverse("view_and_create_transactions")
list_response: Response = api_client.get(
list_transaction_url,
)
assert list_response.status_code == status.HTTP_401_UNAUTHORIZED
Then, we check that if we are logged in we can acccess it.
@pytest.mark.django_db
def test_get_transactions_with_token(api_client: APIClient, user_factory):
email: str = "bobby@email.com"
password: str = "smith"
user = user_factory(email=email, password=password)
list_transaction_url: str = reverse("view_and_create_transactions")
api_client.force_authenticate(user=user)
list_response: Response = api_client.get(
list_transaction_url,
)
assert list_response.status_code == status.HTTP_200_OK
Then let’s test that, if there are transaction in the database under our user, we can see them but we cannot see the ones of a different user.
@pytest.mark.django_db
def test_get_transactions_with_token_actual_transactions(
api_client: APIClient, user_factory, transaction_factory
):
email = "bobby@email.com"
password = "smith"
user = user_factory(email=email, password=password)
_ = transaction_factory()
_ = transaction_factory(user=user)
list_transaction_url: str = reverse("view_and_create_transactions")
api_client.force_authenticate(user=user)
list_response: Response = api_client.get(
list_transaction_url,
)
# only 1 should be available since the other one was created by a
# different user
assert (len(list_response.data)) == 1
assert list_response.status_code == status.HTTP_200_OK
# let's create another category
_ = transaction_factory(user=user)
# now there should be two returened
list_response: Response = api_client.get(
list_transaction_url,
)
assert (len(list_response.data)) == 2
assert list_response.status_code == status.HTTP_200_OK
Let’s check that if we try to post, we are not allowed if we are not authenticated.
@pytest.mark.django_db
def test_post_transaction_without_authentication(api_client):
list_transaction_url: str = reverse("view_and_create_transactions")
data = {
"name": "lala", "amount": "120.23", "amount_currency": "XYZ", "retailer": "Ret", "catgory": "cat", "date": "2022-05-23", "transaction_type": "E", "recurring": "false", "user": 1,
}
list_response: Response = api_client.post(
list_transaction_url, data=data, format="json"
)
assert list_response.status_code == status.HTTP_401_UNAUTHORIZED
What about if we are authenticated? Well, then it should work, and it does.
@pytest.mark.django_db
def test_post_transaction_with_authentication(api_client, user_factory, category_factory, retailer_factory):
initial_count: int = Transaction.objects.count()
# let's create a category and a retailer
user = user_factory()
category: Category = category_factory(slug="unique_category", user=user)
retailer: Retailer = retailer_factory(slug="unique_retailer", user=user)
api_client.force_authenticate(user=user)
list_transaction_url: str = reverse("view_and_create_transactions")
data = {
"name": "lala", "amount": "120.23", "amount_currency": "XYZ", "retailer": "unique_retailer", "category": "unique_category", "date": "2022-05-23", "transaction_type": "E", "recurring": "false", "user": 1,
}
list_response: Response = api_client.post(
list_transaction_url, data=data, format="json"
)
print(list_response.json())
final_count: int = Transaction.objects.count()
assert list_response.status_code == status.HTTP_201_CREATED
assert final_count == initial_count + 1
Finally, we shouldn’t be able to post the same thing twice.
@pytest.mark.django_db
def test_post_transaction_twice(api_client, user_factory, category_factory, retailer_factory):
initial_count: int = Transaction.objects.count()
# let's create a category and a retailer
user = user_factory()
category: Category = category_factory(slug="unique_category", user=user)
retailer: Retailer = retailer_factory(slug="unique_retailer", user=user)
api_client.force_authenticate(user=user)
list_transaction_url: str = reverse("view_and_create_transactions")
data = {
"name": "lala", "amount": "120.23", "amount_currency": "XYZ", "retailer": "unique_retailer", "category": "unique_category", "date": "2022-05-23", "transaction_type": "E", "recurring": "false", "user": 1,
}
list_response: Response = api_client.post(
list_transaction_url, data=data, format="json"
)
# now let's post it again
second_list_response: Response = api_client.post(
list_transaction_url, data=data, format="json"
)
json_response: dict[str, list[str]] = second_list_response.json()
assert "The fields name, amount, amount_currency, retailer, category, date, transaction_type, recurring, user must make a unique set." in json_response["non_field_errors"]
final_count: int = Transaction.objects.count()
assert second_list_response.status_code == status.HTTP_400_BAD_REQUEST
# let's also assert that the count hasn't changed
assert Transaction.objects.count() == final_count
And this is the end of this looong post :) In the next one, we will work on the detail endpoints for Retailer, Category and Transaction so we can update/delete/read individual transactions. Afterwards, we will do something more exciting (which I haven’t thought of yet….)
The commit for this post is here.