api is connected and ready to fight

This commit is contained in:
Ivanov Matvey 2025-01-24 15:59:00 +10:00
parent 8161d77084
commit 9de2aefb66
12 changed files with 143 additions and 7 deletions

View File

@ -1,3 +1,5 @@
# api # api
Describe your project here. ```.env
IMEI_API_TOKEN=
```

View File

@ -8,6 +8,9 @@ authors = [
dependencies = [ dependencies = [
"fastapi[standard]>=0.115.7", "fastapi[standard]>=0.115.7",
"pydantic>=2.10.5", "pydantic>=2.10.5",
"pydantic-settings>=2.7.1",
"requests>=2.32.3",
"aiohttp>=3.11.11",
] ]
readme = "README.md" readme = "README.md"
requires-python = ">= 3.12" requires-python = ">= 3.12"

View File

@ -1,7 +1,11 @@
from typing import Union
from fastapi import APIRouter, status from fastapi import APIRouter, status
from src.schemas import CheckingInput, CheckingOutput from src.service import Checking_imei_service, Authentication_service
from src.schemas import CheckingInput, CheckingOutput, CheckingReport, DeviceProperties
from src.utils.exceptions import AccessDenied
checking_router = APIRouter() checking_router = APIRouter()
@ -9,7 +13,13 @@ checking_router = APIRouter()
@checking_router.post( @checking_router.post(
"/check-imei", "/check-imei",
status_code=status.HTTP_200_OK, status_code=status.HTTP_200_OK,
response_model=CheckingOutput response_model=Union[CheckingOutput | dict]
) )
async def check_imei(body: CheckingInput): async def check_imei(body: CheckingInput) -> Union[DeviceProperties | dict]:
return CheckingOutput(result="OK") if not await Authentication_service.token_is_valid(token=body.token):
raise AccessDenied("token_is_not_valid")
report: Union[CheckingReport | dict] = await Checking_imei_service().request_imei_info(body.imei)
if type(report) is dict:
return report
else:
return report.properties

View File

@ -1,5 +1,7 @@
from fastapi import APIRouter, FastAPI from fastapi import APIRouter, FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse
from src.utils.exceptions import AccessDenied
from src.api import checking_router from src.api import checking_router
@ -22,3 +24,9 @@ main_app_router.include_router(checking_router, tags=["Check IMEI"])
app.include_router(checking_router) app.include_router(checking_router)
@app.exception_handler(AccessDenied)
async def login_exception_handler(request: Request, exc: AccessDenied):
return JSONResponse(
status_code=401,
content={},
)

View File

@ -1,2 +1,5 @@
from .checking import CheckingInput as CheckingInput from .checking import CheckingInput as CheckingInput
from .checking import CheckingOutput as CheckingOutput from .checking import CheckingOutput as CheckingOutput
from .external_checking import CheckingRequest as CheckingRequest
from .external_checking import CheckingReport as CheckingReport
from .external_checking import DeviceProperties as DeviceProperties

View File

@ -1,6 +1,8 @@
from typing import Any from typing import Any
from pydantic import BaseModel, ConfigDict from pydantic import BaseModel, ConfigDict
class CheckingOutput(BaseModel): class CheckingOutput(BaseModel):
model_config = ConfigDict(from_attributes=True) model_config = ConfigDict(from_attributes=True)

View File

@ -0,0 +1,55 @@
from typing import Optional, Union
from pydantic import BaseModel, ConfigDict
class CheckingRequest(BaseModel):
model_config = ConfigDict(from_attributes=True)
deviceId: str
serviceId: int
class CheckingReport(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: str
type: str
status: str
orderId: Optional[str] # in test mode is null
service: "ServiceInfo"
amount: str
deviceId: str
processedAt: int
properties: Union["DeviceProperties", dict] # can be empty dict
class DeviceProperties(BaseModel):
model_config = ConfigDict(from_attributes=True)
deviceName: str
image: str
imei: str
meid: Optional[str] = None # parameter isn't exist in documentation
imei2: Optional[str] = None # parameter isn't exist in documentation
serial: Optional[str] = None # parameter isn't exist in documentation
estPurchaseDate: int
simLock: Optional[bool] = None # parameter is exist only in documentation
warrantyStatus: str
repairCoverage: Union[bool, str] # I got bool, in documentation - str
technicalSupport: Union[bool, str] # I got bool, in documentation - str
replacement: Optional[bool] = None # parameter isn't exist in documentation
demoUnit: bool
refurbished: bool
purchaseCountry: str
fmiOn: bool
lostMode: Union[bool, str] # I got bool, in documentation - str
loaner: bool # parameter isn't exist in documentation
usaBlockStatus: Optional[str] = None # parameter is exist only in documentation
network: Optional[str] = None # parameter is exist only in documentation
class ServiceInfo(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
title: str

View File

@ -0,0 +1,2 @@
from .check_imei import Checking_imei_service as Checking_imei_service
from .authentication import Authentication_service as Authentication_service

View File

@ -0,0 +1,5 @@
class Authentication_service:
@staticmethod
async def token_is_valid(token: str) -> bool:
return True

View File

@ -0,0 +1,35 @@
from aiohttp import ClientSession
from src.settings import settings
from src.schemas import CheckingRequest, CheckingReport
class Checking_imei_service:
def __init__(self):
self.service_url = "https://api.imeicheck.net/v1/checks"
self.service_id = 12 # TODO: change service id after company response
self.headers = {
"Authorization": f"Bearer {settings.IMEI_API_TOKEN}",
"Accept-Language": "en"
}
async def request_imei_info(self, imei: str) -> CheckingReport:
data = CheckingRequest(
deviceId=imei,
serviceId=self.service_id
)
async with ClientSession() as session:
async with session.post(
self.service_url,
headers=self.headers,
data=data.model_dump()
) as response:
try:
raw_response = await response.json()
return CheckingReport(**raw_response)
except Exception as e:
print(type(e), str(e))
# TODO: make good handling of request errors
return dict()

10
api/src/settings.py Normal file
View File

@ -0,0 +1,10 @@
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
IMEI_API_TOKEN: str
settings = Settings()

View File

@ -0,0 +1 @@
class AccessDenied(Exception): ...