Coverage for tests/test_public_typing.py: 100%

1"""Pyright-oriented public API prototypes.""" 

2 

3from __future__ import annotations 

4 

5from contextlib import AbstractAsyncContextManager 

6from datetime import datetime 

7from pathlib import Path 

8from typing import TYPE_CHECKING, Literal, assert_type 

9 

10from snekql import ( 

11 MISSING, 

12 CurrentTimestamp, 

13 DateTime, 

14 Fetched, 

15 Index, 

16 InsertQuery, 

17 Integer, 

18 Missing, 

19 Model, 

20 Pending, 

21 Predicate, 

22 SelectModelQuery, 

23 SelectTupleQuery, 

24 SelectValueQuery, 

25 Text, 

26 Transaction, 

27 UpdateQuery, 

28 insert, 

29 mariadb, 

30 select, 

31 sqlite, 

32 update, 

33) 

34from snekql.testing import mariadb as testing_mariadb 

35 

36 

37class User[S = Pending](Model[S, "User[Fetched]"]): 

38 """Canonical table model used by public API typing examples.""" 

39 

40 id: User.GenCol[int] = Integer( 

41 primary_key=True, 

42 auto_increment=True, 

43 default=MISSING, 

44 ) 

45 email: User.Col[str] = Text(nullable=False) 

46 status: User.Col[str] = Text(nullable=False, default="active") 

47 created_at: User.GenCol[datetime] = DateTime( 

48 server_default=CurrentTimestamp(), 

49 default=MISSING, 

50 ) 

51 

52 

53class SqliteUser[S = Pending](sqlite.Model[S, "SqliteUser[Fetched]"]): 

54 """SQLite namespace table model used by public API typing examples.""" 

55 

56 id: SqliteUser.GenCol[int] = sqlite.Integer( 

57 primary_key=True, 

58 auto_increment=True, 

59 default=MISSING, 

60 ) 

61 email: SqliteUser.Col[str] = sqlite.Text(nullable=False) 

62 

63 

64class MariadbUser[S = Pending](mariadb.Model[S, "MariadbUser[Fetched]"]): 

65 """MariaDB namespace table model used by public API typing examples.""" 

66 

67 id: MariadbUser.GenCol[int] = mariadb.Integer( 

68 primary_key=True, 

69 auto_increment=True, 

70 default=MISSING, 

71 ) 

72 email: MariadbUser.Col[str] = mariadb.Text(nullable=False) 

73 

74 

75if TYPE_CHECKING: 

76 sqlite_config = sqlite.Config(database=Path("app.db")) 

77 _ = assert_type(sqlite_config, sqlite.Config) 

78 sqlite_index = sqlite.Index(SqliteUser.email) 

79 _ = assert_type(sqlite_index, Index[SqliteUser[Pending]]) 

80 sqlite_user = SqliteUser(email="alice@example.com") 

81 _ = assert_type(sqlite_user, SqliteUser[Pending]) 

82 _ = assert_type( 

83 select(SqliteUser), SelectModelQuery[SqliteUser[Pending], SqliteUser[Fetched]] 

84 ) 

85 

86 _ = assert_type(mariadb.Model.__snekql_backend__, Literal["mariadb"]) 

87 _ = assert_type(sqlite.Model.__snekql_backend__, Literal["sqlite"]) 

88 

89 mariadb_config = mariadb.Config(database="app", user="snekql") 

90 _ = assert_type(mariadb_config, mariadb.Config) 

91 test_server_context = testing_mariadb.temporary_mariadb_server( 

92 reset_database=True, 

93 ) 

94 _ = assert_type( 

95 test_server_context, 

96 AbstractAsyncContextManager[testing_mariadb.TemporaryMariaDBServer], 

97 ) 

98 test_server = testing_mariadb.TemporaryMariaDBServer( 

99 auth="insecure", 

100 database="test", 

101 data_directory=Path("data"), 

102 error_log_path=Path("mariadb.err"), 

103 host=None, 

104 password="", 

105 pid_path=Path("mariadb.pid"), 

106 port=None, 

107 socket_path=Path("mariadb.sock"), 

108 transports=frozenset({"unix_socket"}), 

109 user="root", 

110 ) 

111 _ = assert_type(test_server.config(), mariadb.Config) 

112 

113 async def check_test_server_sql_helper() -> None: 

114 """The public SQL helper is async and returns command output.""" 

115 

116 command_result = await test_server.run_sql("SELECT 1", check=False) 

117 _ = assert_type(command_result, testing_mariadb.MariaDBCommandResult) 

118 _ = assert_type(await test_server.reset_database(), None) 

119 

120 mariadb_index = mariadb.Index(MariadbUser.email) 

121 _ = assert_type(mariadb_index, Index[MariadbUser[Pending]]) 

122 mariadb_user = MariadbUser(email="alice@example.com") 

123 _ = assert_type(mariadb_user, MariadbUser[Pending]) 

124 _ = assert_type( 

125 select(MariadbUser), 

126 SelectModelQuery[MariadbUser[Pending], MariadbUser[Fetched]], 

127 ) 

128 

129 pending_user = User(email="alice@example.com") 

130 _ = assert_type(pending_user, User[Pending]) 

131 _ = assert_type(pending_user.id, int | Missing) 

132 _ = assert_type(pending_user.email, str) 

133 _ = assert_type(pending_user.created_at, datetime | Missing) 

134 

135 def check_fetched_user(fetched_user: User[Fetched]) -> None: 

136 """Fetched-state generated values are narrowed by descriptor overloads.""" 

137 

138 _ = assert_type(fetched_user.id, int) 

139 _ = assert_type(fetched_user.email, str) 

140 _ = assert_type(fetched_user.created_at, datetime) 

141 

142 _ = assert_type(select(User), SelectModelQuery[User[Pending], User[Fetched]]) 

143 _ = assert_type( 

144 select(User.email).where(User.email.eq("alice@example.com")).all(), 

145 SelectValueQuery[User[Pending], str], 

146 ) 

147 _ = assert_type( 

148 select(User.email, User.status), 

149 SelectTupleQuery[User[Pending], str, str], 

150 ) 

151 _ = assert_type(User.email.eq("alice@example.com"), Predicate[User[Pending]]) 

152 _ = assert_type(User.email.ne("alice@example.com"), Predicate[User[Pending]]) 

153 _ = assert_type(User.email.is_null(), Predicate[User[Pending]]) 

154 _ = assert_type(User.email.is_not_null(), Predicate[User[Pending]]) 

155 _ = assert_type(User.email.in_("a@example.com"), Predicate[User[Pending]]) 

156 _ = assert_type( 

157 User.email.not_in("a@example.com", "b@example.com"), 

158 Predicate[User[Pending]], 

159 ) 

160 _ = assert_type(User.email.like("%@example.com"), Predicate[User[Pending]]) 

161 _ = assert_type(User.email.not_like("%@example.com"), Predicate[User[Pending]]) 

162 _ = assert_type( 

163 User.email.eq("alice@example.com") & User.status.eq("active"), 

164 Predicate[User[Pending]], 

165 ) 

166 _ = assert_type(Index(User.email), Index[User[Pending]]) 

167 _ = assert_type(Index(User.email, unique=True), Index[User[Pending]]) 

168 _ = assert_type(insert(pending_user), InsertQuery[User[Pending]]) 

169 _ = assert_type( 

170 update(User).set(User.email.to("new@example.com")), 

171 UpdateQuery[User[Pending]], 

172 ) 

173 

174 async def check_fetch_types(transaction: Transaction) -> None: 

175 """Runtime fetch overloads preserve selected result shapes.""" 

176 

177 _ = assert_type( 

178 await transaction.fetch_all(select(User).all()), 

179 list[User[Fetched]], 

180 ) 

181 _ = assert_type( 

182 await transaction.fetch_all(select(User.email).all()), 

183 list[str], 

184 ) 

185 _ = assert_type( 

186 await transaction.fetch_all(select(User.email, User.status).all()), 

187 list[tuple[str, str]], 

188 ) 

189 _ = assert_type( 

190 await transaction.fetch_one(select(User.email).all()), 

191 str | None, 

192 )