Coverage for chatgpt_proxy / types / types.py: 89%

1# MIT License 

2# 

3# Copyright (c) 2025 Tuomo Kriikkula 

4# 

5# Permission is hereby granted, free of charge, to any person obtaining a copy 

6# of this software and associated documentation files (the "Software"), to deal 

7# in the Software without restriction, including without limitation the rights 

8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 

9# copies of the Software, and to permit persons to whom the Software is 

10# furnished to do so, subject to the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included in all 

13# copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 

21# SOFTWARE. 

22 

23"""Common and shared project type definitions.""" 

24 

25import ipaddress 

26from types import SimpleNamespace 

27from typing import TypeAlias 

28 

29import asyncpg 

30import httpx 

31import openai 

32import sanic 

33 

34from chatgpt_proxy.db import models 

35 

36 

37class Context(SimpleNamespace): 

38 _client: openai.AsyncOpenAI | None 

39 _pg_pool: asyncpg.Pool | None 

40 _http_client: httpx.AsyncClient | None 

41 

42 @property 

43 def client(self) -> openai.AsyncOpenAI: 

44 if self._client is None: 

45 raise RuntimeError("Context client is None") 

46 return self._client 

47 

48 @client.setter 

49 def client(self, value: openai.AsyncOpenAI): 

50 self._client = value 

51 

52 @property 

53 def pg_pool(self) -> asyncpg.Pool: 

54 if self._pg_pool is None: 

55 raise RuntimeError("Context pg_pool is None") 

56 return self._pg_pool 

57 

58 @pg_pool.setter 

59 def pg_pool(self, value: asyncpg.Pool): 

60 self._pg_pool = value 

61 

62 @property 

63 def http_client(self) -> httpx.AsyncClient: 

64 if self._http_client is None: 

65 raise RuntimeError("Context http_client is None") 

66 return self._http_client 

67 

68 @http_client.setter 

69 def http_client(self, value: httpx.AsyncClient): 

70 self._http_client = value 

71 

72 

73class RequestContext(SimpleNamespace): 

74 jwt_game_server_address: ipaddress.IPv4Address | None = None 

75 jwt_game_server_port: int | None = None 

76 _game: models.Game | None = None 

77 

78 @property 

79 def game(self) -> models.Game: 

80 if self._game is None: 

81 raise RuntimeError("RequestContext game is None") 

82 return self._game 

83 

84 @game.setter 

85 def game(self, value: models.Game): 

86 self._game = value 

87 

88 

89App: TypeAlias = sanic.Sanic[sanic.Config, Context] 

90 

91 

92class Request(sanic.Request[App, RequestContext]): 

93 app: App 

94 

95 @staticmethod 

96 def make_context() -> RequestContext: 

97 return RequestContext()