-
-
Notifications
You must be signed in to change notification settings - Fork 6
/
api.py
80 lines (57 loc) · 1.96 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
""" Pydantic and Sqlalchemy models for the database
The following class are made
1. User
2. APIRequest
"""
from typing import Optional
from pydantic import Field
from sqlalchemy import (
Column,
ForeignKey,
String,
func,
)
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from nowcasting_datamodel.models.base import Base_Forecast
from nowcasting_datamodel.models.utils import CreatedMixin, EnhancedBaseModel
########
# 1. User
########
class UserSQL(Base_Forecast):
"""ML model that is being used"""
__tablename__ = "user"
uuid = Column(UUID, primary_key=True, server_default=func.gen_random_uuid())
email = Column(String)
api_request = relationship("APIRequestSQL", back_populates="user")
class User(EnhancedBaseModel):
"""ML model that is being used"""
email: Optional[str] = Field(..., description="The name of the model", index=True)
def to_orm(self) -> UserSQL:
"""Change model to MLModelSQL"""
return UserSQL(
email=self.email,
)
########
# 2. APIRequest
########
class APIRequestSQL(Base_Forecast, CreatedMixin):
"""Information about what API route was called"""
__tablename__ = "api_request"
uuid = Column(UUID, primary_key=True, server_default=func.gen_random_uuid())
url = Column(String)
user_uuid = Column(UUID, ForeignKey("user.uuid"), index=True)
user = relationship("UserSQL", back_populates="api_request")
class APIRequest(EnhancedBaseModel):
"""Information about the input data that was used to create the forecast"""
url: str = Field(..., description="The url that was called")
user: Optional[User] = Field(
None,
description="The user associated with this api call",
)
def to_orm(self) -> APIRequestSQL:
"""Change model to APIRequestSQL"""
return APIRequestSQL(
url=self.url,
user=self.user.to_orm() if self.user is not None else None,
)