which will allow us to build randomly a travel plan for the given persons, and also provide the ability to generate HTML report for upcoming journey.

Guide - Travel App

Let's learn AMSDAL by example.

Throughout this guide, we’ll walk you through the creation of a Travel application, which will allow us to build randomly a travel plan for the given persons, and also provide the ability to generate HTML report for upcoming journey.

Example

The sources of this example app are available on GitHub: Travel App Example

The deployed Demo Travel App is available here: Demo Travel App

Step 1 - Create an application

We assume you have already installed AMSDAL CLI. If not, please follow the installation guide. So let’s create a new project.

Create an application

Open the terminal, go to the directory where you want to create the project and run the following command:

amsdal new TravelApp ./

It will create a directory travel_app with an empty AMSDAL application.

Detailed information about the command and the files it has created can be found in the CLI reference.

Step 2 - Defining the models

The first of all, let’s design ERD (entity relationship diagram) for our models:

As we can see, the app will provide the ability to plan journeys, add to these journeys some people, and book properties.

Now we can define our models’ schemas.

Define the models

We will use AMSDAL CLI to generate these models. Go to the travel_app directory and run the following commands in the terminal:

amsdal generate model Country -attrs "name:string code:string"
amsdal generate model Person -attrs "first_name:string last_name:string dob:string"
amsdal generate model Property -attrs "name:string type:string address:string free_parking:boolean free_wifi:boolean photos:has-many:File"
amsdal generate model Booking -attrs "property:belongs-to:Property date:string nights:number:default=1"
amsdal generate model Journey -attrs "start_date:string end_date:string country:belongs-to:Country persons:has-many:Person equipment:dict:string:number bookings:has-many:Booking"

Perfect, now, we have all defined models and are ready to run the local server to check it.

Note

You can find more details regarding the amsdal generate model command in the CLI reference.

Step 3 - Run local server

AMSDL CLI provides the ability to run a local server to test the application. Let’s run it and check the result. From the root directory of the travel_app run the following command in the terminal:

amsdal migrations new
amsdal migrations apply

Note

These steps must be performed each time changes are made to models (creating, deleting, modifying existing models)

amsdal serve

It will run the local API server, the Swagger documentation of it is available at this link: http://localhost:8080/docs

Note

The first run of amsdal serve will initiate the sign-up flow. Just follow the instruction. See details here

Also, we can use the AMSDAL Console - the web portal that is hosted on https://console.amsdal.com/.

An unauthorized user cannot log in and access the data, so you need to create a new user. We can do this using an .env file:

AMSDAL_ADMIN_USER_EMAIL=admin@amsdal.com
AMSDAL_ADMIN_USER_PASSWORD=adminpassword

But for authorization, you also need to specify a JWT key, which can also be done through an env variable, for example:

AUTH_JWT_KEY=secret

Note

Adding or changing variables in the .env file requires a mandatory restart of the application. (It must be completely stopped and restarted with the amsdal serve command)

Now you can restart the server and try to log in.

Login: admin@amsdal.com
Password: adminpassword
API Url: http://localhost:8080

Now, let’s fill our database with some data.

Step 4 - Fixtures

We will use fixtures in order to fill our database.

First of all, let’s create fixtures for Country. Go to the ./src/models/country/ directory and create the fixtures folder. Then, put there the countries.json file with the following content:

[
  {
    "external_id": "us",
    "name": "United States",
    "code": "US"
  },
  {
    "external_id": "gb",
    "name": "United Kingdom",
    "code": "GB"
  },
  {
    "external_id": "de",
    "name": "Germany",
    "code": "DE"
  }
]

As you can see the name and the code are properties of our Country model. Although, we also have the external_id which is just a unique key that you can use in the fixtures as a reference ID to these records. It can be any unique string.

Now, let’s restart our local server, stop and start again - amsdal serve. Check results:

AMSDAL ConsoleSwagger API

Go to https://console.amsdal.com/Country and you will see there are our three countries:

Go to http://localhost:8080/docs and scroll down to Object List API. Click on Try it out and put Country into class_name field, hit an Execute. You will get JSON response with your countries that were created from fixtures.

Let’s do the same for the Property model, src/models/property/fixtures/properties.json:

[
  {
    "external_id": "7fe17f0a-2c63-4076-b0fb-0a75022c489f",
    "name": "Freehand Los Angeles",
    "type": "Hotel",
    "address": "416 W 8th St",
    "free_parking": true,
    "free_wifi": true,
    "photos": [
      "Property/1.jpeg",
      "Property/2.jpeg",
      "Property/3.jpeg"
    ]
  },
  {
    "external_id": "098e6218-75bf-4728-a844-196884ed6a90",
    "name": "Murray Hill East Suites",
    "type": "Apartment",
    "address": "149 E 39th St",
    "free_parking": false,
    "free_wifi": true,
    "photos": [
      "Property/4.jpeg",
      "Property/5.jpeg"
    ]
  },
  {
    "external_id": "355d1b05-182c-48ca-b792-aba237dafa9b",
    "name": "Hafenhaus Gager",
    "type": "Country House",
    "address": "Hauptstraße 1",
    "free_parking": true,
    "free_wifi": false,
    "photos": [
      "Property/6.jpeg",
      "Property/7.jpeg"
    ]
  }
]

As you can see, these fixtures include the photos. For example, the "Property/6.jpeg" means that it should use the photo located in the fixtures folder of the Property model. So we need to put photos there. Create a new files directory in src/models/property/fixtures/ folder. And put there 7 images with corresponding names:

Now, let’s re-run our local API server and see the results by the following link:

https://console.amsdal.com/Property

Note

📎 If you want to know more about the fixtures, see the following documentation: Fixtures

Tip

If you check how the response is returned via API Objects list for Property class name, you will notice that binary data of images are returned in base64 format prefixed with data:binary;base64,. It's because JSON does not support binary data. It means, if you want to create objects with binary properties via API you need to encode the binary value to base64 and send it in the same format: data:binary;base64, YOUR-BASE64-DATA.

Now, let’s add some business logic, for example, we don’t want to allow to create of a journey with a non-adult person ( a person aged less than 18).

Step 5 - Adding hooks

Hooks allow you to add custom logic to the model's lifecycle. For example, you can add a hook that will be executed before the model is saved to the database.

We will use the pre_create hook on the Journey model in order to achieve this. Run the following command: amsdal generate hook --model Journey pre_create. It will create the following file: src/models/journey/hooks/pre_create.py

Let’s edit it and add the following code:

SyncAsync

from datetime import datetime
from datetime import date
from amsdal_utils.models.data_models.reference import Reference
from amsdal_models.classes.helpers.reference_loader import ReferenceLoader


def pre_create(self) -> None:
    self.validate_persons_age()


def validate_persons_age(self) -> None:
    from models.user.person import Person

    for index, person_reference in enumerate(self.persons or []):
        person: Person

        if isinstance(person_reference, Person):
            person = person_reference
        elif isinstance(person_reference, Reference):
            person = ReferenceLoader(person_reference).load_reference()
        else:
            person = ReferenceLoader(Reference(**person_reference)).load_reference()

        try:
            birthdate = datetime.strptime(person.dob, "%Y-%m-%d")
        except (TypeError, ValueError):
            continue

        today = date.today()
        age = today.year - birthdate.year - ((today.month, today.day) < (birthdate.month, birthdate.day))

        if age < 18:
            raise ValueError(f"{person.first_name} {person.last_name}: Age must be 18 or older")

from datetime import datetime
from datetime import date
from amsdal_utils.models.data_models.reference import Reference


async def apre_create(self) -> None:
    await self.validate_persons_age()


async def validate_persons_age(self) -> None:
    from models.user.person import Person

    for index, person_reference in enumerate(self.persons or []):
        person: Person

        if isinstance(person_reference, Person):
            person = person_reference
        elif isinstance(person_reference, Reference):
            person = await person_reference
        else:
            person = await Reference(**person_reference)

        try:
            birthdate = datetime.strptime(person.dob, "%Y-%m-%d")
        except (TypeError, ValueError):
            continue

        today = date.today()
        age = today.year - birthdate.year - ((today.month, today.day) < (birthdate.month, birthdate.day))

        if age < 18:
            raise ValueError(f"{person.first_name} {person.last_name}: Age must be 18 or older")

Note, we will check the age if only this is a creation, not updating. Also, we assume the DOB is stored in YYYY-MM-DD format.

The important thing here is that all related objects can be passed either as an instance of an object or a reference. In our case, the list of people is the list of references. Therefore, in order to get the actual object by reference we need to construct the Reference object and execute load_reference() method. We can handle all the cases by the following code:

SyncAsync

if isinstance(person_reference, Person):
    person = person_reference
elif isinstance(person_reference, Reference):
    person = ReferenceLoader(person_reference).load_reference()
else:
    person = ReferenceLoader(Reference(**person_reference)).load_reference()

if isinstance(person_reference, Person):
    person = person_reference
elif isinstance(person_reference, Reference):
    person = await person_reference
else:
    person = await Reference(**person_reference)

Now, the person is an instance of the Person model with corresponding properties and values.

Note

📎 If you want to know more about references, see the following documentation: References

Also, note how we raise the validation errors, you can raise built-in ValueError exception.

Since we have made changes to the models we need to create and apply new migrations. Run the following commands:

amsdal migrations new 
amsdal migrations apply  

Let’s restart our API server amsdal serve and create a Person with an age less than 18 and try to add this person to a journey.

AMSDAL ConsoleSwagger API

1. Go to https://console.amsdal.com/Person and click on Create button.
2. Fill the form in, the first name, the last name, and put DOB so our person will have an age less than 18, in format YYYY-MM-DD
3. And click on the Submit So, you should see something like:

1. Go to http://localhost:8000/docs and scroll down to Object Create API, click on Try it out.
2. Fill the class_name field with Person
3. Put the following JSON into request body:

   {
     "first_name": "Kara",
     "last_name": "Lloyd",
     "dob": "2020-05-12"
   }
   

   You should get 201 status code and the created person in response body, something like:

   {
     "first_name": "Kara",
     "last_name": "Lloyd",
     "dob": "2020-05-12",
     "_metadata": {
       "address": {
         "resource": "sqlite_state",
         "class_name": "Person",
         "class_version": "a7087c0f82c0465f8223977c37ef4272",
         "object_id": "3c1eb6f90c8d4a4a870d47c3dafdb207",
         "object_version": "42bbc479be054f4e874bbfcc78771e52"
       },
       "class_schema_reference": {
         "ref": {
           "resource": "sqlite_state",
           "class_name": "ClassObject",
           "class_version": "fd6b5582025146209b675e35455dd32f",
           "object_id": "Person",
           "object_version": "a7087c0f82c0465f8223977c37ef4272"
         }
       },
       ...
     },
     ...
   }
   

   The important here is the value under _metadata.address key. It's the address of the object that we can use as a reference. We will use it just in a moment.

Not let’s try to add this person to a new journey:

AMSDAL ConsoleSwagger API

1. Go to https://console.amsdal.com/Journey and click on the Create button
2. Fill in the start date and the end date in the format YYYY-MM-DD, e.g. 2023-08-01 and 2023-08-10, select any country and add our person. You should see something like: Now, click on Submit, you will see the error message:

1. Got to http://localhost:8000/docs and scroll down to Object Create API, click on Try it out.
2. Fill the class_name field with Journey
3. Put the following JSON as a request body:

   {
     "start_date": "2030-08-01",
     "end_date": "2030-08-10",
     "country": {"ref": COUNTRY_ADDRESS},
     "persons": [{"ref": PERSON_ADDRESS}]
   }
   

4. Before submitting, we need to replace COUNTRY_ADDRESS and PERSON_ADDRESS with actual addresses. The PERSON_ADDRESS you can get from previous request. The COUNTRY_ADDRESS one - from the response of Object List API, just pick _metadata.address of any country and put it here as it is. So you should have something like this:

   {
     "start_date": "2030-08-01",
     "end_date": "2030-08-10",
     "country": {
       "ref": {
         "resource": "sqlite_state",
         "class_name": "Country",
         "class_version": "51c77e43822441709a4d4b45f2f7c51b",
         "object_id": "us",
         "object_version": "a6044862269a48ce9a05627b850ec472"
       }
     },
     "persons": [
       {
         "ref": {
           "resource": "sqlite_state", 
           "class_name": "Person", 
           "class_version": "af7ef113a01a4240856154ef2dacf4db",
           "object_id": "230b2021fa0646a0892b5ee27afe8996",
           "object_version": "31918790b0474aefa26074d1ff7efd8b"    
         }
       }
     ]
   }
   

5. Click on Execute and you will get 400 status code and the following JSON response:

   {
     "detail": "Kara Lloyd: Age must be 18 or older",
     "control": ...
   }
   

Perfect!

Note

📎 If you want to know more about all available hooks, see the following documentation: Generate Hooks

You might notice how the records are displayed in select fields on the AMSDAL Console:

It’s not quite informative, so let’s improve it.

Step 6 - Creating a display name modifier

The modifier, it’s a predefined method that allows you to override some specific built-in methods or properties of your end models. Run the following commands in order to generate the blueprints of the display name modifiers for our models:

amsdal generate modifier --model Person display_name
amsdal generate modifier --model Country display_name

it will create modifiers/display_name.py files in each of our specified models. Let’s open this file for the country model and change it to the following:

@property
def display_name(self) -> str:
    return self.name

The self is our instance of the Country model, so we can access any of the fields of this model. In our case, it’s enough to just return the country name and show it in UI.

Now, let’s change src/models/person/modifiers/display_name.py to the following:

@property
def display_name(self) -> str:
    from datetime import datetime
    from datetime import date

    today = date.today()

    try:
        birthdate = datetime.strptime(self.dob, "%Y-%m-%d")
    except (TypeError, ValueError):
        age = "N/A"
    else:
        age = today.year - birthdate.year - ((today.month, today.day) < (birthdate.month, birthdate.day))

    return f"{self.first_name} {self.last_name} ({age} years old)"

As you can see we access two fields of the Person model and combine them to the string using Python’s f string functionality to have the full name of a person.

Let's not forget to make the migrations:

amsdal migrations new
amsdal migrations apply

Now, if you will restart the server:

AMSDAL ConsoleSwagger API

And during creation a journey you will see something like this:

And using the Object List API for these models, in JSON response you will get an extra display_name field:

  ...
  {
    "name": "United States",
    "code": "US",
    "display_name": "United States",
    ...,
  },
  ...

Perfect!

Note

📎 If you want to know more about all available modifiers, see the following documentation: Generate Modifiers

You might notice that we already had similar functionality for calculating the age. We did this in the pre_init hook. Let’s keep following the DRY (Don't repeat yourself) principle and move this functionality to the custom property of the Person model.

Step 7 - Creating custom properties

Run the following command to generate the blueprint of the custom property for the Person model:

amsdal generate property --model Person age

It will create src/models/person/properties/age.py file. Open it and change to the following:

@property
def age(self):
    from datetime import datetime
    from datetime import date

    today = date.today()

    try:
        birthdate = datetime.strptime(self.dob, "%Y-%m-%d")
    except (TypeError, ValueError):
        return "N/A"
    else:
        return today.year - birthdate.year - ((today.month, today.day) < (birthdate.month, birthdate.day))

Now, we can use this property in the pre_create hook for Journey model. Open src/models/journey/hooks/pre_create.py and adjust it to the following:

SyncAsync

from amsdal_utils.models.data_models.reference import Reference
from amsdal_models.classes.helpers.reference_loader import ReferenceLoader


def pre_create(self) -> None:
    self.validate_persons_age()


def validate_persons_age(self) -> None:
    from models.user.person import Person

    for index, person_reference in enumerate(self.persons or []):
        person: Person
        if isinstance(person_reference, Person):
            person = person_reference
        elif isinstance(person_reference, Reference):
            person = ReferenceLoader(person_reference).load_reference()
        else:
            person = ReferenceLoader(Reference(**person_reference)).load_reference()

        if person.age == "N/A":
            raise ValueError("Invalid DOB format. Please use YYYY-MM-DD")

        if person.age < 18:
            raise ValueError(
                f"{person.first_name} {person.last_name}: Age must be 18 or older"
            )

from amsdal_utils.models.data_models.reference import Reference


async def apre_create(self) -> None:
    await self.validate_persons_age()


async def validate_persons_age(self) -> None:
    from models.user.person import Person

    for index, person_reference in enumerate(self.persons or []):
        person: Person

        if isinstance(person_reference, Person):
            person = person_reference
        elif isinstance(person_reference, Reference):
            person = await person_reference
        else:
            person = await Reference(**person_reference)

        if person.age == "N/A":
            raise ValueError("Invalid DOB format. Please use YYYY-MM-DD")

        if person.age < 18:
            raise ValueError(
                f"{person.first_name} {person.last_name}: Age must be 18 or older"
            )

It’s much clear now, is it?

Let’s do the same for the disaply_name modifier of the Person model. Open src/models/person/modifiers/display_name.py and change it to the following:

@property
def display_name(self) -> str:
    return f"{self.first_name} {self.last_name} ({self.age} years old)"

Don't forget to create migrations and restart the server and make sure everything is working correctly.

Perfect! Our custom properties will now be displayed as a separate column, for example for Person we now have an “Age” column. Now it is much more readable!

Note

📎 If you want to know more about the custom properties, see the following documentation: Generate Custom Properties

Now, let’s create a transaction that will plan our journey.

Step 8 - Creating transactions

General

We will use the Transaction feature. The transaction is a kind of function that can combine complex business logic. It can accept the parameters and return some results. In this document, we will create two transactions:

• BuildJourney - this transaction will generate a journey and book hotels. We will learn how to build complex logic and how to use the Model creation in practice.
• GenerateReport - this transaction will generate a report based on the upcoming journey. We will learn how to use the QuerySets, how to add to the application external python libraries.

The journey builder transaction

This transaction will generate a journey and book hotels. In the real world, it would generate it based on the trends or other criteria, although for simplification we will generate it just randomly.

So, this transaction will accept the following params:

• countries - list of possible countries we want to visit.
• nights - how many nights at each stay
• total_nights - how long the whole journey will take e
• persons - list of persons who will attend this journey
• equipment - key-value for storing some equipment and quantity for this journey

Let’s generate a transaction blueprint by running the following command:

amsdal generate transaction BuildJourney

It will create the src/transactions/build_journey.py file, open and change it to the following:

SyncAsync

import random
from datetime import date, timedelta
from datetime import datetime

from amsdal_data.transactions import transaction
from models.user.booking import Booking
from models.user.country import Country
from models.user.journey import Journey
from models.user.person import Person
from models.user.property import Property


@transaction
def BuildJourney(
    countries: list[Country],
    nights: int,
    total_nights: int,
    persons: list[Person],
    equipment: dict[str, int],
):
    country = get_country_by_trend(countries)
    start_date = get_best_start_date(country)
    bookings = book_best_properties(country, start_date, nights, total_nights)
    last_booking = bookings[-1]
    end_date = datetime.strptime(last_booking.date, "%Y-%m-%d") + timedelta(
        days=last_booking.nights
    )
    journey_start_date = start_date.strftime("%Y-%m-%d")
    journey_end_date = end_date.strftime("%Y-%m-%d")

    journey = Journey(
        start_date=journey_start_date,
        end_date=journey_end_date,
        country=country,
        persons=persons,
        equipment=equipment,
        bookings=bookings,
    )
    journey.save()

    return {
        "start_date": journey.start_date,
    }


def get_country_by_trend(countries: list[Country]) -> Country:
    if len(countries) == 0:
        raise ValueError("No countries specified")

    # TODO: get the most trending country for the current year
    return random.choice(countries)


def get_best_start_date(country: Country) -> date:
    # TODO: here we can check a weather forecast for the country and find the best closest start date
    return date.today() + timedelta(days=random.randint(30, 45))


def book_best_properties(
    country: Country,
    start_date: date,
    nights: int,
    total_nights: int,
):
    # TODO: here we can use external API to find the best properties for the country and book the available ones
    bookings = []
    properties = Property.objects.all().execute()
    rest_nights = total_nights

    if len(properties) == 0:
        raise ValueError(
            f"No properties found for the country: {country.name}"
        )

    while len(properties) > 0 and rest_nights > 0:
        _property = random.choice(properties)
        properties.remove(_property)

        if not len(properties) or rest_nights <= 2 * nights:
            book_nights = rest_nights
            rest_nights = 0
        else:
            book_nights = nights
            rest_nights -= nights

        booking = Booking(
            property=_property,
            date=start_date.strftime("%Y-%m-%d"),
            nights=book_nights,
        )
        bookings.append(booking)

    return bookings

import random
from datetime import date, timedelta
from datetime import datetime

from amsdal_data.transactions import async_transaction
from models.user.booking import Booking
from models.user.country import Country
from models.user.journey import Journey
from models.user.person import Person
from models.user.property import Property


@async_transaction
async def BuildJourney(
    countries: list[Country],
    nights: int,
    total_nights: int,
    persons: list[Person],
    equipment: dict[str, int],
):
    country = get_country_by_trend(countries)
    start_date = get_best_start_date(country)
    bookings = await book_best_properties(country, start_date, nights, total_nights)
    last_booking = bookings[-1]
    end_date = datetime.strptime(last_booking.date, "%Y-%m-%d") + timedelta(
        days=last_booking.nights
    )
    journey_start_date = start_date.strftime("%Y-%m-%d")
    journey_end_date = end_date.strftime("%Y-%m-%d")

    journey = Journey(
        start_date=journey_start_date,
        end_date=journey_end_date,
        country=country,
        persons=persons,
        equipment=equipment,
        bookings=bookings,
    )
    await journey.asave()

    return {
        "start_date": journey.start_date,
    }


def get_country_by_trend(countries: list[Country]) -> Country:
    if len(countries) == 0:
        raise ValueError("No countries specified")

    # TODO: get the most trending country for the current year
    return random.choice(countries)


def get_best_start_date(country: Country) -> date:
    # TODO: here we can check a weather forecast for the country and find the best closest start date
    return date.today() + timedelta(days=random.randint(30, 45))


async def book_best_properties(
    country: Country,
    start_date: date,
    nights: int,
    total_nights: int,
):
    # TODO: here we can use external API to find the best properties for the country and book the available ones
    bookings = []
    properties = await Property.objects.all().aexecute()
    rest_nights = total_nights

    if len(properties) == 0:
        raise ValueError(
            f"No properties found for the country: {country.name}"
        )

    while len(properties) > 0 and rest_nights > 0:
        _property = random.choice(properties)
        properties.remove(_property)

        if not len(properties) or rest_nights <= 2 * nights:
            book_nights = rest_nights
            rest_nights = 0
        else:
            book_nights = nights
            rest_nights -= nights

        booking = await Booking(
            property=_property,
            date=start_date.strftime("%Y-%m-%d"),
            nights=book_nights,
        ).asave()
        bookings.append(booking)

    return bookings

Let’s see what we do there. First of all, the async def BuildJourney(...) is our transaction, and as you can see it accepts all arguments that we described before.

Note

⚠️ The important thing here is to annotate all arguments of transactions with types. It helps to properly understand and accept them when you use API to execute these transactions (see POST to /transactions/{transactionname} in the Swagger documentation).

All other functions (e.g. get_country_by_trend, get_best_start_date, etc.) are just helpers that we use inside our transaction. You can put the whole logic just inside the BuildJourney function, although splitting it into smaller functions provides a better understanding of the code.

The idea of this transaction is:

1. it gets one of the passed countries, the most popular country for this year. Although for simplification we just take the country randomly.
2. After that, we resolve the best start date of the journey. For example, we can check the weather forecast for resolved country and pick the warmest day. Although for simplification, we just take it randomly as well.
3. Then we search the most popular properties for this country and for journey dates and book them. Again, we can use some 3rd-party integration to find the best properties, although for simplification we just take them all from our pre-defined database. Note, we loop through our properties and book one by one until we will book a number of nights equal to provided total_nights argument.
4. And when we have a list of bookings, we create our Journey object and put there all referenced objects.

This transaction demonstrates how can we build the complex business logic and create records and store them in DB. As you can see, in order to create and store an object you need just create an instance of this object.

Note

📎 If you want to know more about object creation and updating see the following documentation: Model instances

Let’s now build one more transaction that will build a report.

Report generation transaction

Let’s build the transaction that will generate a report about the upcoming journey in HTML format, and also will include the history of any changes related to it.

Assume we want to generate the following layout for our HTML report:

The business logic will be the following:

1. Our transaction function will not accept arguments
2. It will take the first upcoming journey and collect the stored information from DB
3. Then we will put this information into an HTML template and render the HTML page
4. After that, we will just write this HTML to a file system, although it can be sent via E-mail in the real-world project.

First of all, we need to extend our Journey model and add the start_timestamp field with the number type to have the ability to filter and order our journeys easily. Open src/models/journey/model.json and add this property just next to the start_date:

{
    "title": "Journey",
    "type": "object",
    "control": {
        "type": "object_latest"
    },
    "column_format": {
        "headerTemplate": "StringTemplate",
        "cellTemplate": "ObjectDisplayNameTemplate"
    },
    "properties": {
        "start_date": {
            "title": "start_date",
            "type": "string"
        },
        "start_timestamp": {
            "title": "start_timestamp",
            "type": "number"
        },
        "end_date": {
            "title": "end_date",
            "type": "string"
        },
        "country": {
            "title": "country",
            "type": "Country"
        },
        "persons": {
            "title": "persons",
            "type": "array",
            "items": {
                "type": "Person"
            }
        },
        "equipment": {
            "title": "equipment",
            "type": "dictionary",
            "items": {
                "key_type": "string",
                "value_type": "number"
            }
        },
        "bookings": {
            "title": "bookings",
            "type": "array",
            "items": {
                "type": "Booking"
            }
        }
    },
    "required": [],
    "indexed": []
}

Now, let’s add functionality that will set this timestamp automatically based on the start_date value. We will do that in post_init hook, so run amsdal generate hook --model Journey post_init, open generated post_init.py file, and change it to the following:

def post_init(self, is_new_object, kwargs):
    from datetime import datetime

    try:
        start_date = datetime.strptime(self.start_date, "%Y-%m-%d")
    except (TypeError, ValueError):
        self.start_timestamp = None
    else:
        self.start_timestamp = start_date.timestamp()

This hook will be executed after the instance is initialized and validation was proceed, so we can access instance properties via self.

After these changes, you need to make migrations:

amsdal migrations new
amsdal migrations apply

Now, let’s run the following command to generate a new transaction blueprint:

amsdal generate transaction GenerateReport

Open the src/transactions/generate_report.py file and change it to this:

SyncAsync

from datetime import datetime
import base64
import jinja2
from amsdal_utils.models.data_models.metadata import Metadata
from amsdal_utils.models.enums import Versions
from amsdal_data.transactions import transaction
from amsdal_models.errors import ValueError
from models.user.booking import Booking
from models.user.journey import Journey
from amsdal_utils.models.data_models.reference import Reference
from models.user.person import Person
from amsdal_models.classes.model import Model


@transaction
def GenerateReport():
    upcoming_journey = get_upcoming_journey()

    if not upcoming_journey:
        raise ValueError("No upcoming journey found")

    history_changes = get_history_changes(upcoming_journey)
    html_buffer = render_html(upcoming_journey, history_changes)

    # Send this HTML in E-mail, although we will just write to the file
    with open("report.html", "wt") as f:
        f.write(html_buffer)

    return {
        "html": html_buffer,
    }

from datetime import datetime
import base64
import jinja2
from amsdal_utils.models.data_models.metadata import Metadata
from amsdal_utils.models.enums import Versions
from amsdal_data.transactions import async_transaction
from amsdal_models.errors import ValueError
from models.user.booking import Booking
from models.user.journey import Journey
from amsdal_utils.models.data_models.reference import Reference
from models.user.person import Person
from amsdal_models.classes.model import Model


@async_transaction
async def GenerateReport():
    upcoming_journey = await get_upcoming_journey()

    if not upcoming_journey:
        raise ValueError("No upcoming journey found")

    history_changes = await get_history_changes(upcoming_journey)
    html_buffer = render_html(upcoming_journey, history_changes)

    # Send this HTML in E-mail, although we will just write to the file
    with open("report.html", "wt") as f:
        f.write(html_buffer)

    return {
        "html": html_buffer,
    }

This is not yet ready for running because we need to implement each of these internal functions, although it gives us an understanding of the transaction business logic:

1. So first of all, we need to get the upcoming journey from DB upcoming_journey = get_upcoming_journey()
2. Then we check that we have a journey, otherwise, we raise the validation error
3. After that, we get the history changes for this journey history_changes = get_history_changes(upcoming_journey)
4. Then we render the HTML page html_buffer = render_html(upcoming_journey, history_changes)
5. And finally, we write this HTML to the file system

Let’s start implementing the first inner function - get_upcoming_journey():

SyncAsync

def get_upcoming_journey():
    qs = Journey.objects.filter(start_timestamp__gt=datetime.now().timestamp())
    qs = qs.order_by("start_timestamp")

    return qs.first().execute()

async def get_upcoming_journey():
    qs = Journey.objects.filter(start_timestamp__gt=datetime.now().timestamp())
    qs = qs.order_by("start_timestamp")

    return await qs.first().aexecute()

Just put this function next to our GenerateReport() function as we did it in the previous transaction. As you can see, we build a QuerySet and first of all filter Journeys that have start_timestamp greater than the current timestamp.

After that, we add ordering to this QuerySet by qs.order_by("start_timestamp") which means we need to order our records by the start_timestamp field in the ASC direction.

Each applied function to the QuerySet, e.g. filter, order_by, first, etc., will return a new QuerySet object. So the QuerySet is a kind of immutable object, therefore we override the qs variable after applying the ordering. The first will just take the first record from the ordered list.

Note

📎 If you want to know more about QuerySets, see the following documentation: QuerySets

OK, now let’s implement the next function - get_history_changes(upcoming_journey). This function should return the list of changes, for example, in the following format:

[
    {"date": "2023-04-10", "model": "Person", "action": "changed", "display_name": "Julian Lange"},
    ...
]

So we later can render the list of changes in our PDF report. The implementation of this function can be the following:

SyncAsync

def get_history_changes(journey: Journey):
    history = []

    # get all history changes for the journey itself
    history.extend(_get_history_changes_for_model(Journey, journey, "Journey"))

    # get all history changes for the bookings
    for booking in journey.bookings:
        history.extend(_get_history_changes_for_model(Booking, booking, "Booking"))

    # get all history changes for the persons
    for person in journey.persons:
        history.extend(_get_history_changes_for_model(Person, person, "Person"))

    # sort history by order_key
    history.sort(key=lambda x: x["order_key"])

    return history


def _get_history_changes_for_model(model, obj: Model | Reference, model_name):
    history = []

    qs = model.objects.filter(
        _address__class_version=Versions.ALL,
        _address__object_id=obj.object_id
        if isinstance(obj, Model)
        else obj.ref.object_id,
        _address__object_version_id=Versions.ALL,
    )

    item: Model
    for item in qs.execute():
        history.append(
            {
                "order_key": item.get_metadata().updated_at,
                "date": _ms_to_date(item.get_metadata().updated_at),
                "model": model_name,
                "action": _resolve_action(item.get_metadata()),
                "display_name": getattr(
                    item, "display_name", str(item.get_metadata().address)
                ),
            }
        )

    return history


def _ms_to_date(ms: int):
    return datetime.fromtimestamp(ms / 1000).strftime("%Y-%m-%d")


def _resolve_action(metadata: Metadata):
    if metadata.is_deleted:
        return "Deleted"
    elif metadata.next_version:
        return "Changed"
    else:
        return "Created"

async def get_history_changes(journey: Journey):
    history = []

    # get all history changes for the journey itself
    history.extend(await _get_history_changes_for_model(Journey, journey, "Journey"))

    # get all history changes for the bookings
    for booking in journey.bookings:
        history.extend(await _get_history_changes_for_model(Booking, booking, "Booking"))

    # get all history changes for the persons
    for person in journey.persons:
        history.extend(await _get_history_changes_for_model(Person, person, "Person"))

    # sort history by order_key
    history.sort(key=lambda x: x["order_key"])

    return history


async def _get_history_changes_for_model(model, obj: Model | Reference, model_name):
    history = []

    qs = model.objects.filter(
        _address__class_version=Versions.ALL,
        _address__object_id=obj.object_id
        if isinstance(obj, Model)
        else obj.ref.object_id,
        _address__object_version_id=Versions.ALL,
    )

    item: Model
    for item in await qs.aexecute():
        history.append(
            {
                "order_key": item.get_metadata().updated_at,
                "date": _ms_to_date(item.get_metadata().updated_at),
                "model": model_name,
                "action": _resolve_action(item.get_metadata()),
                "display_name": getattr(
                    item, "display_name", str(item.get_metadata().address)
                ),
            }
        )

    return history


def _ms_to_date(ms: int):
    return datetime.fromtimestamp(ms / 1000).strftime("%Y-%m-%d")


def _resolve_action(metadata: Metadata):
    if metadata.is_deleted:
        return "Deleted"
    elif metadata.next_version:
        return "Changed"
    else:
        return "Created"

As you can see we added a few more internal function-helpers just to get rid of code duplication.

Also, you can see we use the _address__class_version=Versions.ALL QuerySet filter in order to fetch all versions of classes and the _address__object_version_id=Versions.ALL to fetch all versions of the object itself.

Now, let’s write the render_html(upcoming_journey, history_changes) function implementation. We will store the HTML template separately as a static file. Therefore, let’s, first of all, create the directory static in the src one and put there our template.html:

<!DOCTYPE html>
<html>
<head>
    <style>
        body {
            margin: 0;
            font-family: Arial, sans-serif;
            font-size: 12px;
            padding: 10px 20px;
        }

        h1 {
            text-align: center;
        }

        h3 {
            text-align: center;
            font-size: 14px;
            margin-left: 15px;
            margin-top: 35px;
        }

        .amazing-journey {
            flex-grow: 1;
            text-align: right;
        }

        .rows {
            flex-direction: row;
            display: flex;
        }

        .text {
            margin: 15px 5px;
        }

        .text > div {
            margin: 8px 0;
        }

        ul {
            padding-left: 10px;
            margin-top: 0;
        }

        .sub-header {
            text-align: center;
            margin: -10px 0 10px;
        }

        img {
            display: block;
            margin: 5px auto;
        }
    </style>
</head>
<body>
<h1>Journey plan</h1>
<br/>

<div class="rows">
    <div class="rows">
        <div class="text">
            <div>Start date:</div>
            <div>End date:</div>
        </div>
        <div class="text">
            <div><strong>{{journey.start_date}}</strong></div>
            <div><strong>{{journey.end_date}}</strong></div>
        </div>
    </div>
    <h3 class="amazing-journey">
        Amazing journey in <strong>{{ journey.country.display_name }}</strong>
    </h3>
</div>
<div class="rows">
    <div class="text">
        <div>Persons:</div>
        <div>Equipment:</div>
    </div>
    <div class="text">
        <div>
            {% for person in journey.persons %}
            <strong>{{ person.first_name }} {{ person.last_name}}</strong> ({{ person.age }} yo){% if not
            loop.last %}, {% endif %}
            {% endfor %}
        </div>
        <div>
            <ul>
                {% for equipment, quantity in journey.equipment | items %}
                <li><strong>{{ equipment }}</strong> - <strong>{{ quantity }}</strong> quantity</li>
                {% endfor %}
            </ul>
        </div>
    </div>
</div>
{% for booking in journey.bookings %}
<div class="section">
    <h3 class="property">
        <strong>{{ booking.nights|int }} nights</strong> in <strong>{{ booking.property.name }}</strong>
    </h3>
    <div class="sub-header">{{ booking.date }}</div>
    {% for photo in booking.property.photos %}
    <img src="{{ photo.data | base64 }}" height="200px"/>
    {% endfor %}
</div>
{% endfor %}
<div class="history">
    <h3>History of changes</h3>
    {% for item in history_changes %}
    <div class="rows">
        <div class="text">
            <strong>{{ item.date }}</strong>
        </div>
        <div class="text">
            {{ item.model }} "{{ item.display_name | default('N/A', true) }}" was {{ item.action }}.
        </div>
    </div>
    {% endfor %}
</div>

</body>
</html>

In order to render this template we will use the Jinja2 Python library. So we need to add this package to our project and install it. Add the following line (just below of amsdal[cli]) into the requirements.txt file located in the root of our project:

Jinja2

Note, it’s important to have the exact requirements.txt file name for all our dependencies, so do not rename it.

Now install it using pip install -r requirements.txt command.

Ok, now let’s implement our render_html(upcoming_journey, history_changes) function (just add it at the bottom of the src/transactions/generate_report.py file):

def render_html(journey: Journey, history_changes: list):
    loader = jinja2.FileSystemLoader(searchpath="./static")
    env = jinja2.Environment(loader=loader, extensions=["jinja2.ext.loopcontrols"])

    def base64_encode(value):
        if not isinstance(value, bytes):
            raise TypeError("Input must be bytes")

        result = base64.b64encode(value).decode("utf-8")

        return f"data:image/jpeg;base64,{result}"

    env.filters["base64"] = base64_encode

    template = env.get_template("template.html")

    return template.render(
        journey=journey.model_dump(),
        history_changes=history_changes,
    )

As you can see, here we just initiate Jinja2 classes specifies searchpath="./static" where our template is located, and render the template by passing there two variables - journey and history_changes. We also add a custom Jinja2 filter - base64 just in order to display the image in HTML <img src="..."/> tag. As you can see, images are stored in base64 as bytes, so we just decode them to string value.decode("utf-8") and build the string f"data:image/jpeg;base64,{result}" that we put to the IMG tag.

Done! The whole src/transactions/generate_report.py file should be the following:

SyncAsync

from datetime import datetime
import base64
import jinja2
from amsdal_utils.models.data_models.metadata import Metadata
from amsdal_utils.models.enums import Versions
from amsdal_data.transactions import transaction
from models.user.booking import Booking
from models.user.journey import Journey
from amsdal_utils.models.data_models.reference import Reference
from models.user.person import Person
from amsdal_models.classes.model import Model


@transaction
def GenerateReport():
    upcoming_journey = get_upcoming_journey()

    if not upcoming_journey:
        raise ValueError("No upcoming journey found")

    history_changes = get_history_changes(upcoming_journey)
    html_buffer = render_html(upcoming_journey, history_changes)

    # Send this HTML in E-mail, although we will just write to the file
    with open("report.html", "wt") as f:
        f.write(html_buffer)

    return {
        "html": html_buffer,
    }


def get_upcoming_journey():
    qs = Journey.objects.filter(start_date__gt=datetime.now().strftime("%Y-%m-%d"))
    qs = qs.order_by("start_date")

    return qs.first().execute()


def get_history_changes(journey: Journey):
    history = []

    # get all history changes for the journey itself
    history.extend(_get_history_changes_for_model(Journey, journey, "Journey"))

    # get all history changes for the bookings
    for booking in journey.bookings:
        history.extend(_get_history_changes_for_model(Booking, booking, "Booking"))

    # get all history changes for the persons
    for person in journey.persons:
        history.extend(_get_history_changes_for_model(Person, person, "Person"))

    # sort history by order_key
    history.sort(key=lambda x: x["order_key"])

    return history


def _get_history_changes_for_model(model, obj: Model | Reference, model_name):
    history = []

    qs = model.objects.filter(
        _address__class_version=Versions.ALL,
        _address__object_id=obj.object_id
        if isinstance(obj, Model)
        else obj.ref.object_id,
        _address__object_version_id=Versions.ALL,
    )

    item: Model
    for item in qs.execute():
        history.append(
            {
                "order_key": item.get_metadata().updated_at,
                "date": _ms_to_date(item.get_metadata().updated_at),
                "model": model_name,
                "action": _resolve_action(item.get_metadata()),
                "display_name": getattr(
                    item, "display_name", str(item.get_metadata().address)
                ),
            }
        )

    return history


def _ms_to_date(ms: int):
    return datetime.fromtimestamp(ms / 1000).strftime("%Y-%m-%d")


def _resolve_action(metadata: Metadata):
    if metadata.is_deleted:
        return "Deleted"
    elif metadata.next_version:
        return "Changed"
    else:
        return "Created"


def render_html(journey: Journey, history_changes: list):
    loader = jinja2.FileSystemLoader(searchpath="./static")
    env = jinja2.Environment(loader=loader, extensions=["jinja2.ext.loopcontrols"])

    def base64_encode(value):
        if not isinstance(value, bytes):
            raise TypeError("Input must be bytes")

        result = base64.b64encode(value).decode("utf-8")

        return f"data:image/jpeg;base64,{result}"

    env.filters["base64"] = base64_encode

    template = env.get_template("template.html")

    return template.render(
        journey=journey.model_dump(),
        history_changes=history_changes,
    )

from datetime import datetime
import base64
import jinja2
from amsdal_utils.models.data_models.metadata import Metadata
from amsdal_utils.models.enums import Versions
from amsdal_data.transactions import async_transaction
from models.user.booking import Booking
from models.user.journey import Journey
from amsdal_utils.models.data_models.reference import Reference
from models.user.person import Person
from amsdal_models.classes.model import Model


@async_transaction
async def GenerateReport():
    upcoming_journey = await get_upcoming_journey()

    if not upcoming_journey:
        raise ValueError("No upcoming journey found")

    history_changes = await get_history_changes(upcoming_journey)
    html_buffer = render_html(upcoming_journey, history_changes)

    # Send this HTML in E-mail, although we will just write to the file
    with open("report.html", "wt") as f:
        f.write(html_buffer)

    return {
        "html": html_buffer,
    }


async def get_upcoming_journey():
    qs = Journey.objects.filter(start_date__gt=datetime.now().strftime("%Y-%m-%d"))
    qs = qs.order_by("start_date")

    return await qs.first().aexecute()


async def get_history_changes(journey: Journey):
    history = []

    # get all history changes for the journey itself
    history.extend(await _get_history_changes_for_model(Journey, journey, "Journey"))

    # get all history changes for the bookings
    for booking in journey.bookings:
        history.extend(await _get_history_changes_for_model(Booking, booking, "Booking"))

    # get all history changes for the persons
    for person in journey.persons:
        history.extend(await _get_history_changes_for_model(Person, person, "Person"))

    # sort history by order_key
    history.sort(key=lambda x: x["order_key"])

    return history


async def _get_history_changes_for_model(model, obj: Model | Reference, model_name):
    history = []

    qs = model.objects.filter(
        _address__class_version=Versions.ALL,
        _address__object_id=obj.object_id
        if isinstance(obj, Model)
        else obj.ref.object_id,
        _address__object_version_id=Versions.ALL,
    )

    item: Model
    for item in await qs.execute():
        history.append(
            {
                "order_key": item.get_metadata().updated_at,
                "date": _ms_to_date(item.get_metadata().updated_at),
                "model": model_name,
                "action": _resolve_action(item.get_metadata()),
                "display_name": getattr(
                    item, "display_name", str(item.get_metadata().address)
                ),
            }
        )

    return history


def _ms_to_date(ms: int):
    return datetime.fromtimestamp(ms / 1000).strftime("%Y-%m-%d")


def _resolve_action(metadata: Metadata):
    if metadata.is_deleted:
        return "Deleted"
    elif metadata.next_version:
        return "Changed"
    else:
        return "Created"


def render_html(journey: Journey, history_changes: list):
    loader = jinja2.FileSystemLoader(searchpath="./static")
    env = jinja2.Environment(loader=loader, extensions=["jinja2.ext.loopcontrols"])

    def base64_encode(value):
        if not isinstance(value, bytes):
            raise TypeError("Input must be bytes")

        result = base64.b64encode(value).decode("utf-8")

        return f"data:image/jpeg;base64,{result}"

    env.filters["base64"] = base64_encode

    template = env.get_template("template.html")

    return template.render(
        journey=journey.model_dump(),
        history_changes=history_changes,
    )

Let’s run our local server and check it - amsdal serve

AMSDAL ConsoleSwagger API

1. Go to the list of persons and create a Person with an age greater than 18.
2. Now go to transactions and execute the BuildJourney transaction.
3. After that go and run our GenerateReport transaction.

1. Use the Object Create API to create a Person with an age greater than 18 (see examples above).
2. Go to the Transaction Execute API click on Try it out, put the BuildJourney into transaction_name field and the following JSON as request body:

   {
     "countries": [
       {
         "ref": COUNTRY_ADDRESS
       }
     ],
     "nights": 7,
     "total_nights": 15,
     "persons": [
       {
         "ref": PERSON_ADDRESS
       }
     ],
     "equipment": {
       "ski": 1,
       "ski pass": 2
     }
   }
   

   Replace COUNTRY_ADDRESS and PERSON_ADDRESS with actual addresses and hit Exectue. You should get a 200 status code and start_date in JSON response.
3. Now, let's execute the second transaction, change the transaction_name to GenerateReport. This transaction doesn't accept any arguments so put {} as a request body, hit the Execute. You will get a 200 status code and HTML source of generated report.

Now, check the root directory of your project, you should see the report.html file. Try to open it in a browser you should see something like this:

Now you know how to add 3rd party libraries to the AMSDAL application, add static files, and use the QuerySet to get historical data.

Step 9 - Deploy to cloud

Once you have completed the sign-up flow, you are able to deploy your application to AMSDAL Cloud. In order to deploy it just run the following command from the root directory of your application and follow the instructions:

amsdal cloud deploys new

Just after that you will get the URL of your future app, and the deploying process will be run. The URL will be something like: https://main-1265b57653fd458ebdde6ffba78d3b8d.cloud.amsdal.com

The swagger documentation will be accessible by /docs path, e.g. https://main-1265b57653fd458ebdde6ffba78d3b8d.cloud.amsdal.com/docs

In order to check the status of the deploying use the following command:

amsdal cloud deploys

As soon as status will be changed from deploying to deployed you can use the URL of your app to access it. Although, sometimes it can still return 503 HTTP status, that means deploying was complete although the server is still booting. So you need just to wait a bit.

Note

See details about the amsdal cloud deploys command here: AMSDAL CLI - deploy