> ## Documentation Index
> Fetch the complete documentation index at: https://developers.stellar.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Ingest events published from a contract

> Use Stellar RPC's getEvents method for querying events, with a 7 day retention window

Stellar RPC provides a [`getEvents` method] which allows you to query events from a smart contract. However, the data retention window for these events is 7 days at most. If you need access to a longer-lived record of these events you'll want to "ingest" the events as they are published, maintaining your own record or database as events are ingested.

There are many strategies you can use to ingest and keep the events published by a smart contract. Among the simplest might be using a community-developed tool such as [Mercury](https://mercurydata.app) which will take all the infrastructure work off your plate for a low subscription fee. Or, [indexers](https://developers.stellar.org/docs/data/indexers) can provide ingested event data for affordable prices.

Another approach we'll explore here is using a cron job to query Stellar RPC periodically and store the relevant events in a locally stored SQLite database. We are going to use an Object Relational Mapper (ORM), allowing us to write database query directly in Python or JavaScript.

## Setup

> [tab] Python
>
> In a [virtual environment](https://docs.python.org/3/tutorial/venv.html), install the Python dependencies:
>
> ```sh theme={"theme":{"light":"github-light","dark":"github-dark"}}
> pip install sqlalchemy stellar-sdk
> ```

## Setup the Database Client

> [tab] Python
>
> To access the database, we will use [SQLAlchemy](https://www.sqlalchemy.org), which is a frequently used Python library to query database.
>
> We are going to ingest events in a table named `StellarEvent`. In SQLAlchemy, this translates into a class, also called a database model:
>
> ```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
> from typing import Any
> from sqlalchemy import orm, JSON
>
>
> class Base(orm.DeclarativeBase):
>     # needed to tell SQLAlchemy to translate a dictionary into a JSON entry
>     type_annotation_map = {
>         dict[str, Any]: JSON,
>     }
>
>
> class Event(Base):
>     __tablename__ = "StellarEvent"
>
>     id: orm.Mapped[int] = orm.mapped_column(primary_key=True)
>     contract_id: orm.Mapped[str]
>     ledger: orm.Mapped[int]
>     topics: orm.Mapped[dict[str, Any]]
>     value: orm.Mapped[str]
> ```
>
> We will use an in-memory-only SQLite database for this guide, but thanks to the use of an ORM, we could be using any other supported database. We would simply need to change the connection string.
>
> ```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
> from sqlalchemy import create_engine
> engine = create_engine("sqlite://", echo=True)
>
> # the following creates the table in the DB
> Base.metadata.create_all(engine)
> ```
>
> > [!TIP]
> >
> > By setting `echo=True` we can understand what is happening on the database. Creating the database table leads to the following logs:
>
> ```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
> BEGIN (implicit)
> PRAGMA main.table_info("StellarEvent")
> ...
> PRAGMA temp.table_info("StellarEvent")
> ...
> CREATE TABLE StellarEvent (
> 	id INTEGER NOT NULL,
>     contract_id VARCHAR NOT NULL,
>     ledger INTEGER NOT NULL,
> 	topics JSON NOT NULL,
> 	value VARCHAR NOT NULL,
> 	PRIMARY KEY (id)
> )
> ...
> COMMIT
> ```

> [tab] JavaScript
>
> The finer details of choosing a Prisma configuration are beyond the scope of this document. You can get a lot more information in the [Prisma quickstart](https://www.prisma.io/docs/getting-started/quickstart). Here is our Prisma schema's model:
>
> ```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
> model StellarEvent {
>   id          String @id
>   type        String
>   ledger      Int
>   contract_id String
>   topic_1     String?
>   topic_2     String?
>   topic_3     String?
>   topic_4     String?
>   value       String
> }
> ```

> [!NOTE]
>
> Using a database model is very convenient as it allows us to control the database schema programmatically. If we need to change the schema, by adding a new columns for instance, then using an ORM allows us to use very powerful migration tools.

We'll use this model to create and query for the events stored in our database.

## Query Events from Stellar RPC

First, we'll need to query the events from Stellar RPC. This simple example makes an RPC request using the [`getEvents` method], filtering for all `transfer` events that are emitted by the native XLM contract.

> [!NOTE]
>
> We are making some assumptions here. We'll assume that your contract sees enough activity, and that you are querying for events frequently enough that you aren't in danger of needing to figure out the oldest ledger Stellar RPC is aware of. The approach we're taking is to find the largest (most recent) ledger sequence number in the database and query for events starting there. Your use-case may require some logic to determine what the latest ledger is, and what the oldest ledger available is, etc.

> [tab] Python
>
> If we start from scratch, there is no known ledger so we can try to ingest roughly the last 7 days assuming a ledger closes every 6s.
>
> ```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
> import stellar_sdk
>
> soroban_server = stellar_sdk.SorobanServer("https://soroban-testnet.stellar.org")
> ledger = soroban_server.get_latest_ledger().sequence - int(3600 / 6 * 24 * 7)
> ```
>
> Later on, we will be able to start from the latest ingested ledger by making a query to our DB.
>
> ```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
> with orm.Session(engine) as session:
>     stmt = sqlalchemy.select(Event.ledger).where(
>         Event.contract_id == contract_id
>     ).order_by(Event.ledger.desc())
>     ingested_ledger = session.scalars(stmt).first()
>
>     if ingested_ledger:
>       ledger = ingested_ledger
> ```
>
> Let's get events from Stellar RPC!
>
> ```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
> from stellar_sdk.soroban_rpc import EventFilter, EventFilterType
>
> contract_id = "CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC"
>
> res = soroban_server.get_events(
>     ledger,
>     filters=[
>         EventFilter(
>             type=EventFilterType.CONTRACT,
>             contract_ids=[contract_id],
>             topics=[["AAAADwAAAAh0cmFuc2Zlcg==", "*", "*", "*"]],
>         )
>     ],
> )
> events = res.events
> ```

> [tab] JavaScript
>
> We use the `@stellar/stellar-sdk` library:
>
> ```javascript theme={"theme":{"light":"github-light","dark":"github-dark"}}
> import { Server } from "@stellar/stellar-sdk/rpc";
> import { PrismaClient } from "@prisma/client";
>
> const server = new Server("https://soroban-testnet.stellar.org");
> const prisma = new PrismaClient();
>
> let latestEventIngested = await prisma.stellarEvent.findFirst({
>   orderBy: [
>     {
>       ledger: "desc",
>     },
>   ],
> });
>
> let { events } = await server.getEvents({
>   startLedger: latestEventIngested.ledger,
>   filters: [
>     {
>       type: "contract",
>       contractIds: ["CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC"],
>       topics: [["AAAADwAAAAh0cmFuc2Zlcg==", "*", "*", "*"]],
>     },
>   ],
> });
> ```

## Store Events in the Database

Now, we'll check if the `events` object contains any new events we should store, and we do exactly that. We're storing the event's topics and values as base64-encoded strings here, but you could decode the necessary topics and values into the appropriate data types for your use-case.

> [tab] Python
>
> > [!TIP]
> >
> > Make your life easier with a SQLAlchemy [`sessionmaker`](https://docs.sqlalchemy.org/en/20/orm/session_basics.html#using-a-sessionmaker) when making transactions (e.g. add more than one record in a single database call.)
>
> ```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
> import sqlalchemy
> from sqlalchemy.orm import sessionmaker
>
> Session = sessionmaker(engine)
> with Session.begin() as session:
>     events_ = []
>     for event in events:
>         topic_ = event.topic
>         events_.append(Event(contract_id=contract_id, ledger=event.ledger, topics=topic_, value=event.value))
>     session.add_all(events_)
> ```
>
> ```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
> BEGIN (implicit)
> INFO sqlalchemy.engine.Engine COMMIT
> INFO sqlalchemy.engine.Engine BEGIN (implicit)
> INFO sqlalchemy.engine.Engine INSERT INTO "StellarEvent" (contract_id, ledger, topics, value) VALUES (?, ?, ?, ?) RETURNING id
> INFO sqlalchemy.engine.Engine [...] ('CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC', 3311, '["AAAADwAAAAh0cmFuc2Zlcg==", "AAAAEgAAAAAAAAAAJY16rJOcKxQayCR7ayNA80hW5q1U4ypIGOY7NktBfKU=", "AAAAEgAAAAHXkotywnA8z+r365/0701QSlWouXn8m0UOoshCtNHOYQ==", "AAAADgAAAAZuYXRpdmUAAA=="]', 'AAAACgAAAAAAAAAAAAAAAAAAAGQ=')
> INFO sqlalchemy.engine.Engine INSERT INTO "StellarEvent" (contract_id, ledger, topics, value) VALUES (?, ?, ?, ?) RETURNING id
> INFO sqlalchemy.engine.Engine [...] ('CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC', 3325, '["AAAADwAAAAh0cmFuc2Zlcg==", "AAAAEgAAAAAAAAAAJY16rJOcKxQayCR7ayNA80hW5q1U4ypIGOY7NktBfKU=", "AAAAEgAAAAHXkotywnA8z+r365/0701QSlWouXn8m0UOoshCtNHOYQ==", "AAAADgAAAAZuYXRpdmUAAA=="]', 'AAAACgAAAAAAAAAAAAAAAAAAAGQ=')
> ...
> COMMIT
> ```

> [tab] JavaScript
>
> ```javascript theme={"theme":{"light":"github-light","dark":"github-dark"}}
> if (events?.length) {
>   events.forEach(async (event) => {
>     await prisma.stellarEvent.create({
>       data: {
>         id: event.id,
>         type: event.type,
>         ledger: event.ledger,
>         contract_id: event.contractId!.toString(),
>         topic_1: event.topic[0]?.toXDR('base64') || null,
>         topic_2: event.topic[1]?.toXDR('base64') || null,
>         topic_3: event.topic[2]?.toXDR('base64') || null,
>         topic_4: event.topic[3]?.toXDR('base64') || null,
>         value: event.value.toXDR('base64')
>       },
>     });
>   });
> }
> ```

## Run the Script with Cron

A cron entry is an excellent way to automate this script to gather and ingest events every so often. You could configure this script to run as (in)frequently as you want or need. This example would run the script every 24 hours at 1:14 pm:

> [tab] Python
>
> ```sh theme={"theme":{"light":"github-light","dark":"github-dark"}}
> 14 13 * * * python /absolute/path/to/script.py
> ```

> [tab] JavaScript
>
> ```sh theme={"theme":{"light":"github-light","dark":"github-dark"}}
> 14 13 * * * node /absolute/path/to/script.js
> ```

Here's another example that will run the script every 30 minutes:

> [tab] Python
>
> ```sh theme={"theme":{"light":"github-light","dark":"github-dark"}}
> 30 * * * * python /absolute/path/to/script.py
> ```

> [tab] JavaScript
>
> ```sh theme={"theme":{"light":"github-light","dark":"github-dark"}}
> 30 * * * * node /absolute/path/to/script.js
> ```

[`getEvents` method]: ../../../data/apis/rpc/api-reference/methods/getEvents