Process timezone aware fields (with SQLite) in the backend

[drorata]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from datetime import datetime

from typing import Optional

import sqlalchemy as sa
from loguru import logger
from pydantic import validator
from sqlmodel import Field, SQLModel

SUPPORTED_TIME_FORMATS = [
    "%Y-%m-%d %H:%M:%S %z",
    "%Y-%m-%d %z",
]


class TransactionBase(SQLModel):
    transaction_ts: datetime = Field(
        default=...,
        title="The date of the transaction",
        description="Can be casted from YYYY-MM-DD",
        sa_column=sa.Column(sa.DateTime(timezone=True), nullable=False),
    )


class Transaction(TransactionBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class TransactionCreate(TransactionBase):
    @validator("transaction_ts", pre=True)
    def ensure_dt_format(cls, v):
        if isinstance(v, datetime):
            if v.tzinfo:
                logger.debug(f"Creating DT: {v}")
                return v
            raise ValueError(f"Provided datetime {v} doesn't have a time zone")
        # Assume v is a string and trying to parse it
        for dt_format in SUPPORTED_TIME_FORMATS:
            try:
                dt = datetime.strptime(v, dt_format)
                if not dt.tzinfo:
                    raise ValueError(f"No timezone parsed for DT {v}")
                logger.debug(f"Creating DT: {dt}")
                return dt
            except ValueError:
                pass

        raise ValueError(
            f"{v} - unsupported date format. Available formats are: {SUPPORTED_TIME_FORMATS}"
        )


class TransactionRead(TransactionBase):
    id: int

Description

My objective is to have a datetime field which is consistent in terms of timezone. To that end, I believe that the following points have to be considered:

• Input from the "user" has to include a timezone
• SQLModel/pydantic have to parse the timezone and validate it
• The database in the backend has to "know" how to persist timezone aware inputs.

In the example code above, I present my attempt to tackle this. To that end, I'm trying to follow the recommendations in the multiple-models section of the docs. I'm also trying to use the suggestion in this comment.

The API part is implemented as follows:

from typing import List

from fastapi import FastAPI
from sqlmodel import Session, SQLModel, create_engine, select

from foo_portfolio import models

sqlite_file_name = "../data/protfolio.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/transactions/", response_model=models.TransactionRead)
def create_transaction(transaction: models.TransactionCreate):
    with Session(engine) as session:
        db_transaction = models.Transaction.from_orm(transaction)
        session.add(db_transaction)
        session.commit()
        session.refresh(db_transaction)
        return db_transaction


@app.get("/transactions/", response_model=List[models.TransactionRead])
def read_transactions():
    with Session(engine) as session:
        transactions = session.exec(select(models.Transaction)).all()
        return transactions

The questions...

With the above, I can achieve:

• I can create items and parse their timezone correctly using the FastAPI generated interface.
• The timezone is parsed correctly, however, it seems like the backend (SQLite) in my example, seems to ignore the timezone.

This leaves me with the following questions:

1. Am I doing something wrong? Is there a way to persist the timezone in the DB (assuming SQLite)?
2. If not, it seems like I have one of two options. Convert the input to UTC and store a naive UTC timestamp in the DB. In this case, I probably don't need the sa.Column(sa.DateTime(timezone=True), nullable=False) in the definition of TransactionBase. Is it correct? In this case, I assume I "define" that the content of the transaction_ts is supposed/expected to be UTC.
3. Another option, is to split the timezone aware input into two columns: the timestamp and its timezone. Should I want to choose this approach - I don't know how to do it in SQLModel. Any idea?
4. Lastly, which approach is better?

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.8

Python Version

3.11

Additional Context

No response

[8thgencore]

You are on the right track with your approach to handling timezones. However, you are correct that SQLite does not have native support for storing timezone-aware timestamps. Instead, SQLite will convert the timestamp to its local time before storing it.

If you choose to split the timestamp into two columns, you can define them as separate sa.Column objects in your SQLModel class, like this:

class TransactionBase(SQLModel):
    # other columns here
    transaction_ts: datetime
    transaction_tz: str

In this case, you will need to modify your data validation code to validate both the timestamp and the timezone, and you will need to handle timezones correctly when retrieving the data from the database.

As for which approach is better, it depends on your specific use case. Storing a naive UTC timestamp is simpler, but it requires that your application handles timezones correctly when retrieving the data. Storing the timestamp and timezone separately requires more validation and handling code, but it allows for more flexibility in handling timezone.

[drorata]

Thanks @8thgencore for your answer! Assuming that I want to split the into a timestamp and a timezone: how can I handle an input provided as YYYY-MM-DDT20:45:56+02:00 and split it into two fields? Does it even make sense?

[8thgencore]

@drorata
Yes, it makes sense to split the input into a timestamp and a timezone if you want to preserve the timezone information in the database. To split the input into two fields, you can use Python's built-in datetime module to parse the input and extract the UTC timestamp and timezone information.

Here's an example of how you can split the input into a timestamp and timezone in your TransactionBase model:

from datetime import datetime, timezone, timedelta
import re
from sqlmodel import SQLModel, Column

class TransactionBase(SQLModel):
    transaction_ts: datetime = Column(nullable=False)
    transaction_tz: str = Column(nullable=False)

    def __init__(self, **data):
        timestamp_str, tz_str = self._split_timestamp_tz(data['transaction_ts'])
        timestamp = datetime.fromisoformat(timestamp_str).replace(tzinfo=timezone.utc)
        super().__init__(**{'transaction_ts': timestamp, 'transaction_tz': tz_str})

    @staticmethod
    def _split_timestamp_tz(timestamp_str: str):
        # extract timestamp and timezone information from input string using regex
        match = re.match(r'(?P<timestamp>.+)\+(?P<tz_offset>\d{2}):(?P<tz_offset_min>\d{2})$', timestamp_str)
        if not match:
            raise ValueError("Invalid timestamp format")
        timestamp_str = match.group('timestamp')
        tz_offset = int(match.group('tz_offset'))
        tz_offset_min = int(match.group('tz_offset_min'))
        tz_delta = timedelta(hours=tz_offset, minutes=tz_offset_min)
        tz_str = f"+{tz_offset:02d}:{tz_offset_min:02d}"
        if tz_offset < 0:
            tz_str = f"-{abs(tz_offset):02d}:{tz_offset_min:02d}"
            tz_delta = timedelta(hours=-tz_offset, minutes=tz_offset_min)
        return timestamp_str, tz_str

    def to_dict(self):
        return {"transaction_ts": self.transaction_ts.isoformat(), "transaction_tz": self.transaction_tz}

In this example, we define two columns in the TransactionBase model: transaction_ts for the UTC timestamp and transaction_tz for the timezone offset. In the __init__ method, we split the input timestamp string into its components using a regular expression and parse the UTC timestamp and timezone offset separately. We then store the timestamp as a timezone-aware datetime object with UTC timezone and store the timezone offset as a string in the format +HH:MM or -HH:MM. Finally, we define a to_dict method to convert the model to a dictionary for serialization.

Note that this implementation assumes that the input timestamp is in ISO 8601 format with a timezone offset. If the input format can vary, you may need to modify the regular expression or add additional parsing logic to handle different formats.

[drorata]

Thanks for this snippet. Firstly, I think that the following implementation achieves a similar behavior and would be more pythonic. The main difference is that the parsed transaction_ts has no timezone attached to it. I find it less confusing - otherwise, both transaction_ts has a timezone, and there is timezone information in addition.

from datetime import datetime
from sqlmodel import SQLModel, Column

class TransactionBase(SQLModel):
    transaction_ts: datetime = Column(nullable=False)
    transaction_tz: str = Column(nullable=False)

    def __init__(self, **data):
        timestamp, tz_str = self._split_timestamp_tz(data['transaction_ts'])
        super().__init__(**{'transaction_ts': timestamp, 'transaction_tz': tz_str})

    @staticmethod
    def _split_timestamp_tz(timestamp_str: str):
        # extract timestamp and timezone information from input string using regex
        tz_str = timestamp_str[-6:]
        ts_str = timestamp_str[:-6]
        return datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%S"), tz_str

    def to_dict(self):
        return {"transaction_ts": self.transaction_ts.isoformat(), "transaction_tz": self.transaction_tz}

I'm still not sure how to use this approach as part of a FastAPI code. In particular, I don't understand how to build the classes for the creation and reading. Ultimately, the creation is supposed to accept a single input (i.e. a string representing a timestamp with a timezone). This should be saved to an object that has two fields (naive timestamp and an offset).

This way or the other, I decided to go with UTC. See my code below.

[drorata]

For the sake of completeness, here is an attempt on using UTC timestamps:

from datetime import datetime, timezone
from typing import Optional

from loguru import logger
from pydantic import validator
from sqlmodel import Field, SQLModel

SUPPORTED_TIME_FORMATS = [
    "%Y-%m-%d %H:%M:%S %z",
    "%Y-%m-%d %z",
]


class TransactionBase(SQLModel):
    transaction_utc_ts: datetime = Field(
        default=...,
        title="The UTC timestamp of the transaction",
    )


class Transaction(TransactionBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class TransactionCreate(TransactionBase):
    @validator("transaction_utc_ts", pre=True)
    def ensure_dt_format(cls, v):
        if isinstance(v, datetime):
            if v.tzinfo:
                logger.debug(f"Creating DT: {v}")
                return v.astimezone(timezone.utc)
            raise ValueError(f"Provided datetime {v} doesn't have a time zone")
        # Assume v is a string and trying to parse it
        for dt_format in SUPPORTED_TIME_FORMATS:
            try:
                dt = datetime.strptime(v, dt_format)
                if not dt.tzinfo:
                    raise ValueError(f"No timezone parsed for DT {v}")
                logger.debug(f"Creating DT: {dt}")
                return dt.astimezone(timezone.utc)
            except ValueError:
                pass

        raise ValueError(
            f"{v} - unsupported date format. Available formats are: {SUPPORTED_TIME_FORMATS}"
        )


class TransactionRead(TransactionBase):
    id: int

I guess that the only confusing part here, is that the field name transaction_utc_ts in the TransactionCreate class is actually expecting a timestamp with a timezone. Otherwise, it works rather nicely.