Newer
Older
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
import logging as log
from typing import List
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from pydantic import BaseModel
from sqlalchemy import inspect
from sqlalchemy.orm import Session, class_mapper
from pilab.crud.Util import from_json, update_attrs
from pilab.schemas.Key import KeyType
from pilab.events import meta
log.getLogger('sqlalchemy.engine').setLevel(log.WARNING)
class Key(object):
@staticmethod
def create(db: Session, **kwargs):
db_key = from_json(KeyType, kwargs)
db.add(db_key)
db.flush()
return db_key.id
@staticmethod
def delete(db: Session, key_id: str):#DONE
db_key = db.query(KeyType).filter(
KeyType.id == key_id
).first()
db.delete(db_key)
db.flush()
return
@staticmethod
def get(db: Session, id: str):#DONE
db_key = db.query(KeyType).filter(KeyType.id == id).first()
return meta.Key(**vars(db_key))
@staticmethod
def getAll(db: Session, owner: str = None):#DONE
if owner:
db_keys = db.query(KeyType).filter(KeyType.owner == owner).all()
else:
db_keys = db.query(KeyType).all()
keys = []
for db_key in db_keys:
key = meta.Key(**vars(db_key))
keys.append(key)
return keys