Build a geocoding applicaton with Python
Introduction
When it comes to building business process applications, coordinating all parts of the application from user interaction to API calls can become complex and unreliable.
Temporal helps shield you from these issues by providing reliability and operability.
For more detail on how Temporal can help business process applications, please see Temporal's Chief Product Officer's discussion in The Top 3 Use Cases for Temporal (ft. Temporal PMs).
In this tutorial, you'll build a business process application that asks the user for an API key and an address, and the program will geocode it using Geoapify's API to get a latitude and longitude for the user's address.
Prerequisites
Before starting this tutorial:
- Complete the Hello World tutorial
- Install requests (tested with version 2.32.3)
- Get a Geoapify API key
pip install requests
As mentioned in the Hello World tutorial, please be sure the
Temporal Server is running, using temporal server start-dev.
If you're having trouble, sometimes closing that terminal and
restarting it can help.
Now that you have your environment ready, it's time to build an invincible geocoder.
Develop a Workflow to orchestrate your interactions with the user and API
The Workflow is the side-effect-free function that orchestrates the Activities (which are atomic, have potential for side effects, and are perhaps non-deterministic).
In this case, the actions are (1) getting information from the user and (2) querying the API.
The Workflow needs to orchestrate those by executing them in order.
First, create a new file called workflow.py and add
the following code:
from datetime import timedelta
from temporalio import workflow
# Import activity, passing it through the sandbox without reloading the module
with workflow.unsafe.imports_passed_through():
from activities import get_address_from_user, get_api_key_from_user, get_lat_long, QueryParams
_TIMEOUT_5_MINS=5*60
@workflow.defn
class GeoCode:
@workflow.run
async def run(self) -> list:
api_key_from_user = await workflow.execute_activity(
get_api_key_from_user, start_to_close_timeout=timedelta(seconds=_TIMEOUT_5_MINS)
)
address_from_user = await workflow.execute_activity(
get_address_from_user, start_to_close_timeout=timedelta(seconds=_TIMEOUT_5_MINS)
)
query_params = QueryParams(api_key=api_key_from_user, address=address_from_user)
lat_long = await workflow.execute_activity(
get_lat_long, query_params, start_to_close_timeout=timedelta(seconds=_TIMEOUT_5_MINS)
)
return lat_long
Again, this is where the orchestration is happening. Notice how the Workflow definition has one method (more on that later), and that method has calls to multiple Activities. This method is coordinating all the atomic Activities that are non-deterministic and prone to fail. These Activities are explained in the next section.
The GeoCode class is decorated with @workflow.defn, which
tells Temporal that the class is a Workflow.
The async def run() function is decorated with @workflow.run, which tells
Temporal which method to execute when executing the Workflow. As previously mentioned,
the Workflow can have more than one method, but it needs exactly one of them
to be decorated with @workflow.run.
Inside the calls to workflow.execute_activity(), pass the Activity.
If that Activity takes arguments, then pass those in after the Activity:
workflow.execute_activity(activity, args*, ...). It's
recommended to collapse the arguments into a single one using a dataclass,
which you have here as QueryParams (it will be defined in the next section).
With the skeleton in place, you can now develop the Activities.
Develop Activities to interact with the user and the API
In this section, you'll implement the Activities that interact with the outside world. The Workflow is doing the orchestration, and the Activities are doing the atomic actions.
You'll start small and add the following to a new file called activities.py:
from temporalio import activity
@activity.defn
async def get_api_key_from_user() -> str:
return input("Please give your API key: ")
@activity.defn
async def get_address_from_user() -> str:
return input("Please give an address: ")
These functions get input from the user -- namely the API key and the address -- and return the values to the caller. The caller is the Workflow you made earlier, so these Activities are returning their results to the Workflow.
The @activity.defn decorator is what tells the Temporal client
(explained in a few sections) that this function is a Temporal
Activity.
These Activities are the first two that are
called in the Workflow made earlier. After the Workflow
calls these two, it has the user's API key and address. Next,
it calls an Activity called get_lat_long, with an argument of
type QueryParams. Now is the time to implement those. Add the following
to the activities.py file that you just made:
import requests
from dataclasses import dataclass
@dataclass
class QueryParams:
api_key: str
address: str
@activity.defn
async def get_lat_long(query_params: QueryParams) -> list:
base_url = "https://api.geoapify.com/v1/geocode/search"
params = {
"text": query_params.address,
"apiKey": query_params.api_key
}
response = requests.get(base_url, params=params, timeout=1000)
response_json = response.json()
lat_long = response_json['features'][0]['geometry']['coordinates']
return lat_long
As mentioned before, the argument to workflows and Activities are
recommended to be dataclasses. In this case, the Activity is
an API call that needs the user's location and API key, so you'll
bundle those as a data class. That's QueryParams.
Next, you need to define the Activity. As before, you need to
decorate the function with @activity.defn. The function accepts
a QueryParams object, calls the Geoapify API, and returns the coordinates
from the result. The Workflow receives this resulting latitude and longitude
and assigns the variable response_from_geoapify to them.
The next two steps put these plans into action. You'll make and run a worker, and then you'll run the Workflow you made in section 1.
Create a Worker to host your Workflow and Activities
The last conceptual piece you need is a worker. The worker
is the process that connects to the Temporal Service, and listens
on a certain Task Queue for any work to do. Here is how you can make
a worker factory. Make a new file called make_worker.py and
enter the following
from temporalio.client import Client
from temporalio.worker import Worker
from activities import get_address_from_user, get_api_key_from_user, get_lat_long
from workflow import GeoCode
def make_worker(client: Client):
worker = Worker(
client,
task_queue="geocode-task-queue",
workflows=[GeoCode],
activities=[get_address_from_user, get_api_key_from_user, get_lat_long]
)
return worker
The argument, client, is the connection to the Temporal Service.
The task_queue is the Task Queue that the Worker listens on (later,
when you run the Workflow, you'll put items on that Task Queue). It also
accepts a workflows argument. This is the list of Workflows it can
process. Lastly, it accepts a list of Activities.
Now that you have a way to make a Worker, it's time to connect to the
Temporal Service, use that function to make a Worker, and run the Worker.
You'll do this in a new file called run_worker.py.
import asyncio
from temporalio.client import Client
from make_worker import make_worker
async def main():
client = await Client.connect("localhost:7233", namespace="default")
worker = make_worker(client)
await worker.run()
if __name__ == "__main__":
asyncio.run(main())
This code connects to the Temporal Service using Client.connect.
You'll run this code soon, and when you do, the Temporal Service
will need to be running for this line to work (which is why it was mentioned in the
prerequisites).
Next, the code passes that client into the preceeding make_worker function.
This returns a Worker, which you use to
call the .run() method. This is what makes the Worker run
and start listening for work on the Task Queue.
At this point, you can open a new terminal (keep the service running
in a different terminal). Navigate to the project directory, and run
python run_worker.py (it won't output anything yet).
It will start listening, but it has nothing
to do yet because there is nothing on the queue. And that's exactly
what the next section is for. In it, you will run the Workflow,
which will put tasks on the queue for the Worker.
Run the Workflow to execute the business process
This is the last step. The process will soon run.
The last thing you need to do is execute the Workflow. To do this,
you need to connect to the Temporal Service (again, with Client.connect),
and then call the .execute_workflow() method, as shown later. Enter
the following code in a new file called run_workflow.py.
import asyncio
from workflow import GeoCode
from temporalio.client import Client
async def main():
# Create a client connected to the server at the given address
client = await Client.connect("localhost:7233")
# Execute a workflow
lat_long = await client.execute_workflow(
GeoCode.run, id="geocode-workflow", task_queue="geocode-task-queue"
)
print(f"Lat long: {lat_long}")
if __name__ == "__main__":
asyncio.run(main())
In this piece, you connect to the service, then call execute_workflow.
As you can see, it has a return value that it puts in lat_long. This
return value comes from the return value of the Workflow definition.
The arguments are the following:
- The Workflow method that was decorated with
@workflow.defn. - The ID for the Workflow. This is shown in the WebUI to help you identify which Workflow is which.
- The Task Queue. This is the queue to which the Workflow (and its Activities) are added.
After that, the code prints the return value.
You're ready to run the code. Open a third terminal (the other
two processes should still be running), navigate to the project
directory, and run python run_workflow.py. At this point, the
application is officially running. If you look at the terminal
that's running the Worker (not the terminal that's running the Workflow),
it should be asking you for your
API key. You should enter the Geoapify API key mentioned in
the prerequisites. Next, it will ask you for an address. Try
whatever you want. It will then query the Geoapify API to
geocode the address, and it will return the latitude and longitude
of the address to the terminal running the Workflow.
Conclusion
You have built a business process application that runs invincibly with Temporal.
Next Steps
Now on your own, try defining a Retry Policy and applying it to the Activities.