Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import datetime
- import uuid
- from marshmallow_sqlalchemy import ModelSchema
- from sqlalchemy import Table, Column, ForeignKey, Integer, String, Text, DateTime, Float
- from sqlalchemy import create_engine, text, func
- from sqlalchemy.orm import Session, relationship, backref, joinedload_all, sessionmaker
- from sqlalchemy.ext.mutable import MutableDict
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.dialects.postgresql import HSTORE, ARRAY, UUID
- engine = create_engine('postgresql://vladislav@localhost:5432/postgres')
- Session = sessionmaker(bind=engine)
- session = Session()
- print(session)
- Base = declarative_base()
- class Service(object):
- def fields(self):
- return self.__table__.columns.keys()
- def to_dict(self, skip=[]):
- return {k: getattr(self, k) for k in self.fields() if k not in skip}
- serialize_model = to_dict
- @staticmethod
- def jsonify(o):
- if isinstance(o, (datetime.datetime, datetime.date, datetime.timedelta, uuid.UUID)):
- return str(o)
- if hasattr(o, 'serialize_model'):
- return o.to_dict()
- def __repr__(self):
- fields = ','.join(["{}='{}'".format(k, v) for (k, v) in self.to_dict().items()])
- return "{}({})".format(self.__class__.__name__, fields)
- user_role = Table('user_role', Base.metadata,
- Column('user_id', ForeignKey('users.id'), primary_key=True),
- Column('role_id', ForeignKey('roles.id'), primary_key=True)
- )
- class User(Base, Service):
- __tablename__ = 'users'
- id = Column('id', UUID(as_uuid=True), primary_key=True, server_default=text("uuid_generate_v4()"))
- username = Column('username', String(128), nullable=False, unique=True)
- password = Column('password', String(128), nullable=False)
- registered_at = Column('registered_at', DateTime(), default=func.now())
- last_login_at = Column('last_login_at', DateTime())
- reports = relationship('UserReport', back_populates='owner')
- incidents = relationship('Incident', primaryjoin='User.id==foreign(Incident.user_id)')
- roles = relationship('Role', back_populates='users', secondary=user_role)
- class Role(Base, Service):
- __tablename__ = 'roles'
- id = Column('id', Integer, primary_key=True)
- name = Column('name', String(128), nullable=False, unique=True)
- users = relationship('User', back_populates='roles', secondary=user_role)
- permissions = relationship('Permission', back_populates='roles')
- class Permission(Base, Service):
- __tablename__ = 'permissions'
- id = Column('id', Integer, primary_key=True)
- role_id = Column('role_id', Integer, ForeignKey('roles.id'))
- action = Column('action', String(64), nullable=False)
- resource = Column('resource', String(128), nullable=False)
- roles = relationship('Role', back_populates='permissions')
- class UserReport(Base, Service):
- __tablename__ = 'user_reports'
- id = Column('id', Integer, primary_key=True)
- kind = Column('kind', String(64), nullable=False)
- name = Column('name', String(128), nullable=False)
- parameters = Column('parameters', MutableDict.as_mutable(HSTORE))
- user_id = Column('user_id', UUID(as_uuid=True), ForeignKey('users.id'))
- created_at = Column('created_at', DateTime(), default=func.now())
- owner = relationship('User', back_populates='reports')
- class Person(Base, Service):
- __tablename__ = 'people'
- id = Column('id', Integer, primary_key=True)
- indent = Column('indent', ARRAY(String(128)), server_default=text('array[]::varchar(128)[]'))
- data = Column('data', MutableDict.as_mutable(HSTORE))
- wabsessions = relationship('WABSession', back_populates='person')
- proxies = relationship('PersonProxy', back_populates='person')
- timeframes = relationship('TimeFrame', back_populates='person')
- class PersonProxy(Base, Service):
- __tablename__ = 'person_rights'
- person_id = Column('person_id', Integer, ForeignKey('people.id'), primary_key=True)
- proxy_id = Column('proxy_id', Integer, ForeignKey('proxygw.id'), primary_key=True)
- rights = Column('rights', ARRAY(String(128)), server_default=text('array[]::varchar(128)[]'))
- person = relationship('Person', back_populates='proxies')
- proxy = relationship('ProxyGW', back_populates='people')
- class ProxyGW(Base, Service):
- __tablename__ = 'proxygw'
- id = Column('id', Integer, primary_key=True)
- indent = Column('indent', String(128), nullable=False)
- addr = Column('addr', String(256))
- token = Column('token', String(128))
- username = Column('username', String(32))
- password = Column('password', String(32))
- data = Column('data', MutableDict.as_mutable(HSTORE), server_default=text("hstore('')"), nullable=False)
- people = relationship('PersonProxy', back_populates='proxy')
- timeframes = relationship('TimeFrame', back_populates='proxy')
- class TimeFrame(Base, Service):
- __tablename__ = 'timeframes'
- id = Column('id', Integer, primary_key=True)
- start = Column('start', DateTime(), nullable=False)
- stop = Column('stop', DateTime(), nullable=False)
- person_id = Column('person_id', Integer, ForeignKey('people.id'))
- proxy_id = Column('proxy_id', Integer, ForeignKey('proxygw.id'))
- person = relationship('Person', back_populates='timeframes')
- proxy = relationship('ProxyGW', back_populates='timeframes')
- class WABSession(Base, Service):
- __tablename__ = 'wabsessions'
- id = Column('id', Integer, primary_key=True)
- session_start = Column('session_start', DateTime(), nullable=False)
- session_stop = Column('session_stop', DateTime(), nullable=False)
- wab_addr = Column('wab_addr', String(512), nullable=False)
- session_id = Column('session_id', String(128), nullable=False)
- client_ip = Column('client_ip', String(16), nullable=False)
- target_ip = Column('target_ip', String(16), nullable=False)
- user = Column('user', String(512), nullable=False)
- device = Column('device', String(512), nullable=False)
- service = Column('service', String(16), nullable=False)
- account = Column('account', String(512), nullable=False)
- person_id = Column('person_id', Integer, ForeignKey('people.id'))
- person = relationship('Person', back_populates='wabsessions')
- events = relationship('Event', back_populates='wabsession')
- incidents = relationship('Incident', back_populates='wabsession')
- event_term = Table('event_term', Base.metadata,
- Column('event_id', ForeignKey('events.id'), primary_key=True),
- Column('term_id', ForeignKey('terms.id'), primary_key=True)
- )
- class Event(Base, Service):
- __tablename__ = 'events'
- id = Column('id', Integer, primary_key=True)
- wabsession_id = Column('wabsession_id', Integer, ForeignKey('wabsessions.id'))
- recorded_at = Column('recorded_at', DateTime())
- event_type = Column('event_type', String(512), nullable=False)
- event_data = Column('event_data', MutableDict.as_mutable(HSTORE))
- wabsession = relationship('WABSession', back_populates='events')
- terms = relationship('Term', back_populates='events', secondary=event_term)
- incident = relationship('Incident', back_populates='event')
- class Term(Base, Service):
- __tablename__ = 'terms'
- id = Column('id', Integer, primary_key=True)
- value = Column('value', String(256), nullable=False)
- events = relationship('Event', secondary=event_term, back_populates='terms')
- class Incident(Base, Service):
- __tablename__ = 'incidents'
- id = Column('id', Integer, primary_key=True)
- indent = Column('indent', String(128))
- kind = Column('kind', String(256), nullable=False)
- score = Column('score', Float, nullable=False)
- data = Column('data', String(256))
- severity = Column('severity', String(32))
- # enum LOW, MIDDLE, HIGH, CRITICAL
- state = Column('state', String(32), default='NEW')
- # enum NEW, OPEN, CLOSED, REOPENED
- wabsession_id = Column('wabsession_id', Integer, ForeignKey('wabsessions.id'))
- event_id = Column('event_id', Integer, ForeignKey('events.id'))
- user_id = Column('user_id', UUID(as_uuid=True))
- wabsession = relationship('WABSession', back_populates='incidents')
- event = relationship('Event', back_populates='incident')
- owner = relationship('User', primaryjoin='foreign(Incident.user_id)==User.id')
- class ColorList(Base, Service):
- __tablename__ = 'color_lists'
- id = Column('id', Integer, primary_key=True)
- name = Column('name', String(128), nullable=False)
- factor = Column('factor', Float, nullable=False)
- items = relationship('ColorListItem', back_populates='list')
- class ColorListItem(Base, Service):
- __tablename__ = 'color_list_items'
- id = Column('id', Integer, primary_key=True)
- pattern = Column('pattern', String(256), nullable=False)
- list_id = Column('list_id', Integer, ForeignKey('color_lists.id'))
- list = relationship('ColorList', back_populates='items')
- class TermSchema(ModelSchema):
- class Meta:
- model = Term
- sqla_session = session
- term_schema = TermSchema(Term)
- result = term_schema.validate({'id': 100, 'value': 'qwerty'})
- print(result)
Add Comment
Please, Sign In to add comment