Coverage for tests/test_datetime.py: 100%

1from collections.abc import Callable 9 ctx1abcdefghi

2from datetime import date, datetime, time, timedelta, timezone, tzinfo 9 ctx1abcdefghi

3from io import StringIO 9 ctx1abcdefghi

4from operator import add, sub 9 ctx1abcdefghi

5from typing import Annotated, Any 9 ctx1abcdefghi

7import pytest 9 ctx1abcdefghi

8from alembic.autogenerate import compare_metadata, render_python_code 9 ctx1abcdefghi

9from alembic.migration import MigrationContext 9 ctx1abcdefghi

10from alembic.operations import Operations, ops 9 ctx1abcdefghi

11from pydantic import AwareDatetime, NaiveDatetime, ValidationError 9 ctx1abcdefghi

12from sqlalchemy import Column, Date, DateTime, Engine, Interval, Time 9 ctx1abcdefghi

13from sqlalchemy.dialects import mysql, postgresql, sqlite 9 ctx1abcdefghi

14from sqlalchemy.exc import StatementError 9 ctx1abcdefghi

15from sqlalchemy.schema import CreateTable 9 ctx1abcdefghi

16from sqlmodel import ( 9 ctx1abcdefghi

17 Field,

18 Session,

19 SQLModel,

20 UTCDateTime,

21 create_engine,

22 select,

23 update,

24)

27@pytest.mark.parametrize("nullable", [False, True]) 9 ctx1abcdefghi

28@pytest.mark.parametrize( 9 ctx1abcdefghi

29 "annotation, timezone_enabled, error_type",

30 [

31 (datetime, True, None),

32 (AwareDatetime, True, "timezone_aware"),

33 (NaiveDatetime, False, "timezone_naive"),

34 (Annotated[datetime, AwareDatetime], True, "timezone_aware"),

35 (Annotated[datetime, NaiveDatetime], False, "timezone_naive"),

36 (Annotated[datetime, AwareDatetime()], True, "timezone_aware"),

37 (Annotated[datetime, NaiveDatetime()], False, "timezone_naive"),

38 (Annotated[NaiveDatetime, AwareDatetime], True, "timezone_aware"),

39 (Annotated[AwareDatetime, NaiveDatetime], False, "timezone_naive"),

40 (Annotated[datetime, NaiveDatetime, AwareDatetime], True, "timezone_aware"),

41 (Annotated[datetime, AwareDatetime, NaiveDatetime], False, "timezone_naive"),

42 (Annotated[datetime, "other metadata"], True, None),

43 ],

44)

45def test_datetime_types( 9 ctx1abcdefghi

46 annotation: Any, timezone_enabled: bool, error_type: str | None, nullable: bool

47) -> None:

48 if nullable: 9 ctx1stuvwxyzA

49 annotation = annotation | None 9 ctx1stuvwxyzA

51 class Event(SQLModel, table=True): 9 ctx1stuvwxyzA

52 id: int | None = Field(default=None, primary_key=True) 9 ctx1stuvwxyzA

53 occurred_at: annotation 9 ctx1stuvwxyzA

55 column = SQLModel.metadata.tables["event"].c.occurred_at 9 ctx1stuvwxyzA

56 assert isinstance(column.type, UTCDateTime if timezone_enabled else DateTime) 9 ctx1stuvwxyzA

57 assert column.type.timezone is timezone_enabled 9 ctx1stuvwxyzA

58 assert column.nullable is nullable 9 ctx1stuvwxyzA

60 aware = datetime(2026, 1, 1, 12, tzinfo=timezone(timedelta(hours=2))) 9 ctx1stuvwxyzA

61 naive = datetime(2026, 1, 1, 12) 9 ctx1stuvwxyzA

62 valid = aware if timezone_enabled else naive 9 ctx1stuvwxyzA

63 assert Event.model_validate({"occurred_at": valid}).occurred_at == valid 9 ctx1stuvwxyzA

64 if error_type is None: 9 ctx1stuvwxyzA

65 assert Event.model_validate({"occurred_at": naive}).occurred_at == naive 9 ctx1stuvwxyzA

66 else:

67 invalid = naive if timezone_enabled else aware 9 ctx1stuvwxyzA

68 with pytest.raises(ValidationError) as exc_info: 9 ctx1stuvwxyzA

69 Event.model_validate({"occurred_at": invalid}) 9 ctx1stuvwxyzA

70 assert exc_info.value.errors()[0]["type"] == error_type 9 ctx1stuvwxyzA

71 if nullable: 9 ctx1stuvwxyzA

72 assert Event.model_validate({"occurred_at": None}).occurred_at is None 9 ctx1stuvwxyzA

75def test_inherited_annotated_datetime() -> None: 9 ctx1abcdefghi

76 class EventBase(SQLModel): 9 ctx2kblbmbnbobpbqbrbsb

77 occurred_at: Annotated[datetime, NaiveDatetime, Field(index=True)] 9 ctx2kblbmbnbobpbqbrbsb

79 class Event(EventBase, table=True): 9 ctx2kblbmbnbobpbqbrbsb

80 id: int | None = Field(default=None, primary_key=True) 9 ctx2kblbmbnbobpbqbrbsb

82 column = SQLModel.metadata.tables["event"].c.occurred_at 9 ctx2kblbmbnbobpbqbrbsb

83 assert column.type.timezone is False 9 ctx2kblbmbnbobpbqbrbsb

84 assert column.index is True 9 ctx2kblbmbnbobpbqbrbsb

87@pytest.mark.parametrize("annotation", [datetime, AwareDatetime, NaiveDatetime]) 9 ctx1abcdefghi

88@pytest.mark.parametrize("timezone_enabled", [False, True]) 9 ctx1abcdefghi

89@pytest.mark.parametrize("use_column", [False, True]) 9 ctx1abcdefghi

90def test_explicit_datetime_type( 9 ctx1abcdefghi

91 annotation: Any, timezone_enabled: bool, use_column: bool

92) -> None:

93 sa_type = DateTime(timezone=timezone_enabled) 9 ctx2bbcbdbebfbgbhbibjb

94 if use_column: 9 ctx2bbcbdbebfbgbhbibjb

95 field = Field(sa_column=Column(sa_type)) 9 ctx2bbcbdbebfbgbhbibjb

96 else:

97 field = Field(sa_type=sa_type) 9 ctx2bbcbdbebfbgbhbibjb

99 class Event(SQLModel, table=True): 9 ctx2bbcbdbebfbgbhbibjb

100 id: int | None = Field(default=None, primary_key=True) 9 ctx2bbcbdbebfbgbhbibjb

101 occurred_at: annotation = field 9 ctx2bbcbdbebfbgbhbibjb

102

103 assert SQLModel.metadata.tables["event"].c.occurred_at.type is sa_type 9 ctx2bbcbdbebfbgbhbibjb

104

105

106def test_explicit_datetime_type_class() -> None: 9 ctx1abcdefghi

107 class Event(SQLModel, table=True): 9 ctx2CbDbEbFbGbHbIbJbKb

108 id: int | None = Field(default=None, primary_key=True) 9 ctx2CbDbEbFbGbHbIbJbKb

109 occurred_at: datetime = Field(sa_type=DateTime) 9 ctx2CbDbEbFbGbHbIbJbKb

110

111 assert SQLModel.metadata.tables["event"].c.occurred_at.type.timezone is False 9 ctx2CbDbEbFbGbHbIbJbKb

112

113

114def test_other_temporal_types() -> None: 9 ctx1abcdefghi

115 class Event(SQLModel, table=True): 9 ctx2] ^ _ ` { | } ~ ab

116 id: int | None = Field(default=None, primary_key=True) 9 ctx2] ^ _ ` { | } ~ ab

117 day: date 9 ctx2] ^ _ ` { | } ~ ab

118 at: time 9 ctx2] ^ _ ` { | } ~ ab

119 duration: timedelta 9 ctx2] ^ _ ` { | } ~ ab

120

121 columns = SQLModel.metadata.tables["event"].c 9 ctx2] ^ _ ` { | } ~ ab

122 assert isinstance(columns.day.type, Date) 9 ctx2] ^ _ ` { | } ~ ab

123 assert isinstance(columns.at.type, Time) 9 ctx2] ^ _ ` { | } ~ ab

124 assert isinstance(columns.duration.type, Interval) 9 ctx2] ^ _ ` { | } ~ ab

125

126

127@pytest.mark.parametrize( 9 ctx1abcdefghi

128 "dialect, aware_type, naive_type",

129 [

130 (

131 postgresql.dialect(),

132 "TIMESTAMP WITH TIME ZONE",

133 "TIMESTAMP WITHOUT TIME ZONE",

134 ),

135 (mysql.dialect(), "DATETIME", "DATETIME"),

136 (sqlite.dialect(), "DATETIME", "DATETIME"),

137 ],

138)

139def test_datetime_column_types(dialect: Any, aware_type: str, naive_type: str) -> None: 9 ctx1abcdefghi

140 class Event(SQLModel, table=True): 9 ctx2tbubvbwbxbybzbAbBb

141 id: int | None = Field(default=None, primary_key=True) 9 ctx2tbubvbwbxbybzbAbBb

142 occurred_at: datetime 9 ctx2tbubvbwbxbybzbAbBb

143 local_time: NaiveDatetime 9 ctx2tbubvbwbxbybzbAbBb

144

145 ddl = str(CreateTable(SQLModel.metadata.tables["event"]).compile(dialect=dialect)) 9 ctx2tbubvbwbxbybzbAbBb

146 assert f"occurred_at {aware_type} NOT NULL" in ddl 9 ctx2tbubvbwbxbybzbAbBb

147 assert f"local_time {naive_type} NOT NULL" in ddl 9 ctx2tbubvbwbxbybzbAbBb

148

149

150def test_datetime_round_trip(database_engine: Engine) -> None: 9 ctx1abcdefghi

151 class Event(SQLModel, table=True): 9 ctx1jklmnopqr

152 id: int | None = Field(default=None, primary_key=True) 9 ctx1jklmnopqr

153 occurred_at: datetime 9 ctx1jklmnopqr

154 local_time: NaiveDatetime 9 ctx1jklmnopqr

155 optional_time: datetime | None = None 9 ctx1jklmnopqr

156

157 aware = datetime(2026, 1, 1, 12, tzinfo=timezone(timedelta(hours=2))) 9 ctx1jklmnopqr

158 naive = datetime(2026, 1, 1, 12) 9 ctx1jklmnopqr

159 engine = database_engine 9 ctx1jklmnopqr

160 SQLModel.metadata.create_all(engine) 9 ctx1jklmnopqr

161 with Session(engine) as session: 9 ctx1jklmnopqr

162 event = Event(occurred_at=aware, local_time=naive) 9 ctx1jklmnopqr

163 session.add(event) 9 ctx1jklmnopqr

164 session.flush() 9 ctx1jklmnopqr

165 # Binding doesn't change the original value on the model.

166 assert event.occurred_at is aware 9 ctx1jklmnopqr

167 session.commit() 9 ctx1jklmnopqr

168 session.refresh(event) 9 ctx1jklmnopqr

169 assert event.occurred_at == aware 9 ctx1jklmnopqr

170 assert event.occurred_at.tzinfo is timezone.utc 9 ctx1jklmnopqr

171 assert event.local_time == naive 9 ctx1jklmnopqr

172 assert event.optional_time is None 9 ctx1jklmnopqr

173 event_id = event.id 9 ctx1jklmnopqr

174

175 with Session(engine) as session: 9 ctx1jklmnopqr

176 loaded = session.get(Event, event_id) 9 ctx1jklmnopqr

177 assert loaded is not None 9 ctx1jklmnopqr

178 assert loaded.occurred_at.tzinfo is timezone.utc 9 ctx1jklmnopqr

179 # Query parameters use the same UTC normalization as inserted values.

180 assert ( 9 ctx1jklmnopqr

181 session.exec(select(Event).where(Event.occurred_at == aware)).one()

182 is loaded

183 )

184 assert session.exec(select(Event.occurred_at)).one().tzinfo is timezone.utc 9 ctx1jklmnopqr

185 later = aware + timedelta(hours=1) 9 ctx1jklmnopqr

186 session.exec(update(Event).values(occurred_at=later)) 9 ctx1jklmnopqr

187 session.commit() 9 ctx1jklmnopqr

188 session.refresh(loaded) 9 ctx1jklmnopqr

189 assert loaded.occurred_at == later 9 ctx1jklmnopqr

190 assert loaded.occurred_at.tzinfo is timezone.utc 9 ctx1jklmnopqr

191

192

193class UndefinedOffset(tzinfo): 9 ctx1abcdefghi

194 def utcoffset(self, dt: datetime | None) -> None: 9 ctx1abcdefghi

195 return None 9 ctx1BCDEFGHIJ

196

197

198@pytest.mark.parametrize("offset", [None, UndefinedOffset()]) 9 ctx1abcdefghi

199@pytest.mark.parametrize("operation", ["insert", "update", "filter"]) 9 ctx1abcdefghi

200def test_naive_database_parameters( 9 ctx1abcdefghi

201 database_engine: Engine, offset: tzinfo | None, operation: str

202) -> None:

203 class Event(SQLModel, table=True): 9 ctx1BCDEFGHIJ

204 id: int | None = Field(default=None, primary_key=True) 9 ctx1BCDEFGHIJ

205 occurred_at: datetime 9 ctx1BCDEFGHIJ

206

207 naive = datetime(2026, 1, 1, 12, tzinfo=offset) 9 ctx1BCDEFGHIJ

208 event = Event.model_validate({"occurred_at": naive}) 9 ctx1BCDEFGHIJ

209 # Plain datetime validation remains permissive, including a tzinfo object

210 # that doesn't actually provide an offset.

211 assert event.occurred_at is naive 9 ctx1BCDEFGHIJ

212 engine = database_engine 9 ctx1BCDEFGHIJ

213 SQLModel.metadata.create_all(engine) 9 ctx1BCDEFGHIJ

214 with Session(engine) as session: 9 ctx1BCDEFGHIJ

215 with pytest.raises(StatementError, match="timezone information") as exc_info: 9 ctx1BCDEFGHIJ

216 if operation == "insert": 9 ctx1BCDEFGHIJ

217 session.add(event) 9 ctx1BCDEFGHIJ

218 session.flush() 9 ctx1BCDEFGHIJ

219 elif operation == "update": 9 ctx1BCDEFGHIJ

220 session.exec(update(Event).values(occurred_at=naive)) 9 ctx1BCDEFGHIJ

221 else:

222 session.exec(select(Event).where(Event.occurred_at > naive)) 9 ctx1BCDEFGHIJ

223 assert isinstance(exc_info.value.orig, ValueError) 9 ctx1BCDEFGHIJ

224 assert "NaiveDatetime" in str(exc_info.value.orig) 9 ctx1BCDEFGHIJ

225

226

227@pytest.mark.parametrize("timezone_enabled", [False, True]) 9 ctx1abcdefghi

228def test_sqlite_explicit_datetime_bypasses_utc_processing( 9 ctx1abcdefghi

229 timezone_enabled: bool,

230) -> None:

231 class Event(SQLModel, table=True): 9 ctx1#$%'()*+,

232 id: int | None = Field(default=None, primary_key=True) 9 ctx1#$%'()*+,

233 occurred_at: datetime = Field(sa_type=DateTime(timezone=timezone_enabled)) 9 ctx1#$%'()*+,

234

235 aware = datetime(2026, 1, 1, 12, tzinfo=timezone(timedelta(hours=2))) 9 ctx1#$%'()*+,

236 engine = create_engine("sqlite://") 9 ctx1#$%'()*+,

237 SQLModel.metadata.create_all(engine) 9 ctx1#$%'()*+,

238 with Session(engine) as session: 9 ctx1#$%'()*+,

239 event = Event(occurred_at=aware) 9 ctx1#$%'()*+,

240 session.add(event) 9 ctx1#$%'()*+,

241 session.commit() 9 ctx1#$%'()*+,

242 session.refresh(event) 9 ctx1#$%'()*+,

243 assert event.occurred_at == datetime(2026, 1, 1, 12) 9 ctx1#$%'()*+,

244 assert event.occurred_at.tzinfo is None 9 ctx1#$%'()*+,

245

246

247def test_postgresql_datetime_processing() -> None: 9 ctx1abcdefghi

248 # PostgreSQL drivers receive aware parameters and may return values in the

249 # connection's timezone. Exercise the complete SQLAlchemy type processors.

250 dialect = postgresql.dialect() 9 ctx1KLMNOPQRS

251 type_ = UTCDateTime().dialect_impl(dialect) 9 ctx1KLMNOPQRS

252 bind = type_.bind_processor(dialect) 9 ctx1KLMNOPQRS

253 result = type_.result_processor(dialect, None) 9 ctx1KLMNOPQRS

254 assert bind is not None 9 ctx1KLMNOPQRS

255 assert result is not None 9 ctx1KLMNOPQRS

256 aware = datetime(2026, 1, 1, 12, tzinfo=timezone(timedelta(hours=2))) 9 ctx1KLMNOPQRS

257 bound = bind(aware) 9 ctx1KLMNOPQRS

258 assert bound == aware 9 ctx1KLMNOPQRS

259 assert bound.tzinfo is timezone.utc 9 ctx1KLMNOPQRS

260 loaded = result(aware) 9 ctx1KLMNOPQRS

261 assert loaded == aware 9 ctx1KLMNOPQRS

262 assert loaded.tzinfo is timezone.utc 9 ctx1KLMNOPQRS

263 assert bind(None) is None 9 ctx1KLMNOPQRS

264 assert result(None) is None 9 ctx1KLMNOPQRS

265

266

267@pytest.mark.parametrize( 9 ctx1abcdefghi

268 "operation, reverse",

269 [(add, False), (sub, False), (add, True)],

270)

271def test_postgresql_datetime_interval_arithmetic( 9 ctx1abcdefghi

272 operation: Callable[[Any, Any], Any], reverse: bool

273) -> None:

274 class Event(SQLModel, table=True): 9 ctx1-./:;=?@[

275 id: int | None = Field(default=None, primary_key=True) 9 ctx1-./:;=?@[

276 occurred_at: datetime 9 ctx1-./:;=?@[

277

278 occurred_at = SQLModel.metadata.tables["event"].c.occurred_at 9 ctx1-./:;=?@[

279 duration = timedelta(days=30) 9 ctx1-./:;=?@[

280 expression = ( 9 ctx1-./:;=?@[

281 operation(duration, occurred_at)

282 if reverse

283 else operation(occurred_at, duration)

284 )

285 # Adding or subtracting a duration still produces a UTC datetime.

286 assert isinstance(expression.type, UTCDateTime) 9 ctx1-./:;=?@[

287 # Exercise the actual parameter processor used for database execution.

288 parameter = expression.left if reverse else expression.right 9 ctx1-./:;=?@[

289 dialect = postgresql.dialect() 9 ctx1-./:;=?@[

290 bind = parameter.type.dialect_impl(dialect).bind_processor(dialect) 9 ctx1-./:;=?@[

291 bound = bind(parameter.value) if bind is not None else parameter.value 9 ctx1-./:;=?@[

292 assert bound == duration 9 ctx1-./:;=?@[

293

294

295def test_existing_sqlite_utc_data() -> None: 9 ctx1abcdefghi

296 class Event(SQLModel, table=True): 9 ctx1TUVWXYZ01

297 id: int | None = Field(default=None, primary_key=True) 9 ctx1TUVWXYZ01

298 occurred_at: datetime 9 ctx1TUVWXYZ01

299

300 engine = create_engine("sqlite://") 9 ctx1TUVWXYZ01

301 with engine.begin() as connection: 9 ctx1TUVWXYZ01

302 connection.exec_driver_sql( 9 ctx1TUVWXYZ01

303 "CREATE TABLE event (id INTEGER NOT NULL PRIMARY KEY, "

304 "occurred_at DATETIME NOT NULL)"

305 )

306 connection.exec_driver_sql( 9 ctx1TUVWXYZ01

307 "INSERT INTO event VALUES (1, '2026-01-01 12:00:00.000000')"

308 )

309 context = MigrationContext.configure(connection, opts={"compare_type": True}) 9 ctx1TUVWXYZ01

310 assert compare_metadata(context, SQLModel.metadata) == [] 9 ctx1TUVWXYZ01

311

312 with Session(engine) as session: 9 ctx1TUVWXYZ01

313 event = session.get(Event, 1) 9 ctx1TUVWXYZ01

314 assert event is not None 9 ctx1TUVWXYZ01

315 assert event.occurred_at == datetime(2026, 1, 1, 12, tzinfo=timezone.utc) 9 ctx1TUVWXYZ01

316 assert event.occurred_at.tzinfo is timezone.utc 9 ctx1TUVWXYZ01

317

318

319def test_alembic_datetime_rendering() -> None: 9 ctx1abcdefghi

320 class Event(SQLModel, table=True): 9 ctx123456789!

321 id: int | None = Field(default=None, primary_key=True) 9 ctx123456789!

322 occurred_at: datetime 9 ctx123456789!

323

324 table = SQLModel.metadata.tables["event"] 9 ctx123456789!

325 migration = ops.UpgradeOps(ops=[ops.CreateTableOp.from_table(table)]) 9 ctx123456789!

326 rendered = render_python_code(migration) 9 ctx123456789!

327 assert "sqlmodel.sql.sqltypes.UTCDateTime()" in rendered 9 ctx123456789!

328 # Execute the generated migration, including the custom type constructor.

329 namespace: dict[str, Any] = {} 9 ctx123456789!

330 exec( 9 ctx123456789!

331 "import sqlalchemy as sa\nimport sqlmodel.sql.sqltypes\n"

332 "from alembic import op\ndef upgrade():\n " + rendered,

333 namespace,

334 )

335 output = StringIO() 9 ctx123456789!

336 context = MigrationContext.configure( 9 ctx123456789!

337 dialect_name="postgresql",

338 opts={"as_sql": True, "output_buffer": output},

339 )

340 with Operations.context(context): 9 ctx123456789!

341 namespace["upgrade"]() 9 ctx123456789!

342 assert "occurred_at TIMESTAMP WITH TIME ZONE NOT NULL" in output.getvalue() 9 ctx123456789!

Coverage for tests / test_datetime.py: 100%

216 statements