Fastapi

gistfile1.txt

import time
from urllib.request import Request
from fastapi import FastAPI, requests
from typing import Optional
from pydantic import HashModel
from starlette.background import BackgroundTasks


class License(HashModel):
    licensestartdate: str
    api_calls: int
    total_apis: int
    api_id: int
    licenseenddate: str
app = FastAPI()

@app.get('/License')
def all():
    return [format(pk) for pk in License.all_pks()]




def format(pk: str, License):
    License = License.get(pk)

    return {
        'api_id': License.pk,
        'api_calls': License.api_calls,
        'total_apis': License.total_apis,
        'licensestartdate': License.datetime,
        'licenseenddate':License.licenseenddate,

    }


@app.post('/License')
def create(License: License):
    return License.save()


@app.get('/License/{pk}')
def get(pk: str):
    return License.get(pk)


@app.delete('/License/{pk}')
def delete(pk: str):
    return License.delete(pk)





class additonalapi(HashModel):
    api_id: str
    tries_left: int
    additon_apis: int
    dateextend: str
    status: str  # pending, completed, refunded



@app.get('/addditionapi/{pk}')
def get(pk: str):
    return additonalapi.get(pk)




@app.post('/additionapi')
async def create(request: Request, background_tasks: BackgroundTasks, additonalapi_completed, additionalapi):  # id, quantity
    body = await request.json()

    req = requests.get('http://localhost:8000/License/%s' % body['id'])
    License = req.json()

    additonalapi = additionalapi(
        api_id=License['api_id'],
        total_apis=10 - License['api_id'],
        apicalls= 10000 - License['api_id'],
        additionapi= 500 + License['tries_left'],
        dateextend=body['datetime'],
        status='pending'

    )
    additonalapi.save()

    background_tasks.add_task(additonalapi_completed, additonalapi)

    return additonalapi


def additionapi_completed(additionapi: additonalapi):
    time.sleep(5)
    additionapi.status = 'completed'
    additonalapi.save()


# in deployment 90 days time 10000 api call. If in this api in successful decrement 1 at one time this at 0.
# in other model additonal number increment of+1 1 1











""""
 #
app = FastAPI()


# insertcrete
@app.get("apicalls/")
async def add_apicall():
    apicall = {
        "apiid": "1"
                 "license_start_date": "datetime"
                                       "license_end_date": "datetime"
    }
    json_data = jsonable_encoder(apicall)
    return JSONResponse(content=json_data)
    # total_tries = 10000
    # total_apis = 0

    # if apicalls > total_tries:
    # total_apis = apicalls + 1
    # total_tries = total_apis - 1
    # return apicalls

    # if total_apis > 10000:
    # additionaltries = additionalapis + 1
    # return additionaltries, {"you r using additonal apis"}
    ##   total_apis = apicalls + 1
    #  return apicall


# view
@app.get("/apicalls")
async def get_apicalls():
    return get_apicalls


# retrieveall
@app.get("/apicalls{id}")
async def get_apicalls():
    return get_apicalls


# delete
@app.delete("apicalls/{id}")
async def delete_apicalls():
    return delete_apicalls


# update
@app.put("/apicalls{id}")
async def update_apicalls():
    return update_apicalls

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/LicenseInfo", response_model=List[schemas.LicenseInfo])
def LicenseInfo_action_list(limit: int = 100, offset: int = 0, db: SessionLocal = Depends(get_db)):
    LicenseInfo = crud.list_LicenseInfo(db, offset, limit)
    return LicenseInfo

@app.get("/LicenseInfo/{LicenseInfo}", response_model=schemas.LicenseInfo)
def LicenseInfo_action_retrieve(LicenseInfo: int, db: SessionLocal = Depends(get_db)):
    LicenseInfo = crud.get_LicenseInfo(db, LicenseInfo)
    if LicenseInfo is None:
        raise HTTPException(status_code=404)
    return LicenseInfo

@app.post("/LicenseInfo", response_model=schemas.LicenseInfoCreate)
def LicenseInfo_action_create(data: schemas.LicenseInfoCreate, db: SessionLocal = Depends(get_db)):
    LicenseInfo = crud.create_LicenseInfo(db, data)
    return LicenseInfo


@app.put("/LicenseInfo/{LicenseInfo}", response_model=schemas.LicenseInfo)
def LicenseInfo_action_retrieve(LicenseInfo_id: int, data: schemas.LicenseInfoUpdate,  db: SessionLocal = Depends(get_db)):
    LicenseInfo = crud.update_LicenseInfo(db, LicenseInfo_id, data)
    if LicenseInfo is None:
        raise HTTPException(status_code=404)
    return LicenseInfo


@app.delete("/LicenseInfo/{LicenseInfo}", status_code=204)
def LicenseInfo_action_retrieve(LicenseInfo_id: int,  db: SessionLocal = Depends(get_db)):
    crud.drop_LicenseInfo(db, LicenseInfo)
    return LicenseInfo
#insertcrete
@app.get("apicalls/")
def add_apicall(apiid, days=None, total_apis=None):

    limitdays = 90 * days
    if apiid["total_apis"] > 0:
        total_tries = total_apis - apiid
        apiid["total_tries"] = ((apiid["total_apis"])) - apiid
        # --- =10000 - apiid
        for total_tries in limitdays:
            return apicalls
    else:
        days["total_tries"] < total_apis
        return {"you r adding additonal apis full"}



#view
@app.get("/apicalls")
async def get_apicalls(api_calls=None):
    #response = await apicalls_pydantic.from_tortoise_orm(api_calls.all())
    return {"Status": "ok", "data": response}

#retrieveall
@app.get("/apicalls{id}")
async def get_apicalls(id:int, apiid=None, tries_left=None):
    apiid = await apiid.get(id=id)
    tries_left = await tries_left.tries_left

    #response = await apicalls_pydantic.from_queryset_single(apicalls.get(id=id))
    return {
        "status": "ok"
        # all json
    }

#delete
@app.delete("apicalls/{id}")
async def delete_apicalls(id:int, apicall_pydantic=None, apiid=None, total_tries=None):
    apiid = await apiid.get(id=id)
    total_apis = await total_tries.total_tries
    apiid.delete()

#update
@app.put("/apicalls{id}")
async def update_apicalls(update_api=None):

  update_api["date Published"] = datetime.utcnow()
#if user == owner and update_api[tries_left][add_apicall]:







#def addLicense(LicenseInfo, api_id):


    #api_id = max([a[api_id] for a in api_id]) + 1

   # api = {
        #1: {
         #   "version": "1.1",
        #     "apicalls": "0",
      #      "total_tries": "10000",
     #       "tries_left": "0",
    #        "license_start_date": "joindate",

   #     }
  #  }
 #   return addLicense

#@app.get("get_apis/{api_id}")
#def get_apis(api_id: int):
 #   return api_id


#in deployment 90 days time 10000 api call. If in this api in successful decrement 1 at one time this at 0.
#in other model additonal number increment of+1 1 1

                   #LicenseInfo Schem@s
#insert
#def each(apicall):
 #   pass
#with open('api.json','r') as f:
 #   api = json.load(f)['api']
#print(api)

#@app.get("api/{apiid}", status_code=200)
#def LicenseInfo(apiid: int, api=None):
  #  api = [a for a in api if a['api']==apiid]
 #   return api[0] if len(api) > 0 else {}

    #datetime = date
    #days = date * 90
    #if total_tries in days:
       # each(apicall=total_tries - 1)
      #     print(apicall)
     #      return apicall

    #for Validity in CreateLicenseInfo:
        #if Validity in apicall:
        #apicall == 0
         #   return Validity == False
        #else:
            #return Validity == True

           # return {CreateLicenseInfo}

    #limitdays = 90 * days
    #if apiid["total_apis"] > 0:
        #total_tries = total_apis - apiid
       # apiid["total_tries"] = ((apiid["total_apis"])) - apiid
        # --- =10000 - apiid
      #  for total_tries in limitdays:
     #       return apicalls

    #else:
      #  days["total_tries"] < total_apis
     #   return {"you r adding additonal apis full"}

    #for additionaltries in additionalapis:
    #    additionaltries = +1
   #     tries_left = additionaltries - 1
  #      return additionalapis

#@app.get("Search/", status_code=200)
#def search_api(apiid: Optional[int] = Query(None, title: "Total_tries", description: "The Filter is for apiid"),total_tries:Optional[int]
#= Query(None, title: "Total_tries", description: "The Filter is for total_tries"))
#view
#@app.get('LicenseInfo/')
#def add_License(api : api):
    #api = max([a['apiid']for a in api] + 1)
    #total_tries = 10000
   # tries_left = total_tries - api
  #  tries_left = {
        #"apiid": api.apiid,
        #"tries_left": api.tries_left,
        #"apicalls": api.apiid,
    #}
    #api.append(api)
    #return api

    #return Version, datetime, Validity, license_start_date, license_end_date, total_tries


#retrieve
#@app.get('LicenseInfo/{apiid}')
#def getLicense(apiid : int):
    #license = {
      #  1:{
     #        "license_start_date":"datetime.day",

   #     }

  #  }

 #   return license(apiid)

#delete
#@app.delete('/')

#update item
#@app.put('/')
#def UpdateLicenseInfo(Version, datetime, Validity, license_start_date, license_end_date, total_tries, tries_left, additional_tries):
 #   return ViewLicense()
                    #LOG Schem@s
#insert
#@app.post("apiLogs/{additonltries}")
#def additionapi(Version, datetime, Validity, license_start_date, license_end_date, total_tries, tries_left, additional_tries,
   #             additionaltries=None):
  #  if additionaltries in additionapi:
        additionaltries = +1
        
 #       return additionapi[additionaltries][datetime]

#view
#@app.get("apiLogs/")
#def Viewapi(Version, datetime, Validity, license_start_date, license_end_date, total_tries, tries_left, additional_tries):
 #   return additionapi
#retrieve
#@app.get('/')

#delete
#@app.delete('/')

#update item
#@app.put('/')
#def updateadditionapi(Version, datetime, Validity, license_start_date, license_end_date, total_tries, tries_left, additional_tries,
     #                 additionaltries=None):
    #if additionaltries in tries_left:
      #  tries_left = additionaltries - 1
     #   return additionapi[tries_left][datetime]






"""""