__init__.py
2.44 KB
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
81
82
83
84
85
import logging
from tangsel.tools.this_framework import get_settings
from sqlalchemy import engine_from_config, Column, String, ForeignKey, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import (scoped_session, sessionmaker, relationship, backref)
from zope.sqlalchemy import register
from ..base_table import (
BaseJenisPelayanan, BaseBerkasTerima, BaseBerkasKirim,
BasePstPermohonan, BasePstDetail, BaseSeksi, BaseDatObjekPajak, BaseSppt)
_logging = logging.getLogger(__name__)
session_factory = sessionmaker()
ReadPbbDBSession = scoped_session(session_factory)
register(ReadPbbDBSession)
ReadPbbBase = declarative_base()
pbb_schema = 'pbb'
PBB_ARGS = {'extend_existing': True, 'schema': pbb_schema}
class ReadSeksi(ReadPbbBase, BaseSeksi):
__tablename__ = "ref_seksi"
__table_args__ = (PBB_ARGS,)
class ReadJenisPelayanan(ReadPbbBase, BaseJenisPelayanan):
__tablename__ = "ref_jns_pelayanan"
__table_args__ = (PBB_ARGS,)
class ReadDatObjekPajak(ReadPbbBase, BaseDatObjekPajak):
__tablename__ = "dat_objek_pajak"
__table_args__ = (PBB_ARGS,)
class ReadPstPermohonan(ReadPbbBase, BasePstPermohonan):
__tablename__ = "pst_permohonan"
__table_args__ = (PBB_ARGS,)
class ReadPstDetail(ReadPbbBase, BasePstDetail):
__tablename__ = "pst_detail"
__table_args__ = (PBB_ARGS,)
kd_seksi_berkas = Column(String(2), ForeignKey(ReadSeksi.kd_seksi))
tgl_penyerahan = Column(DateTime)
nip_penyerah = Column(String(18))
seksi_berkas = relationship(ReadSeksi, backref=backref("pst_detail"))
class ReadBerkasTerima(ReadPbbBase, BaseBerkasTerima):
__tablename__ = "berkas_terima"
__table_args__ = (PBB_ARGS,)
class ReadBerkasKirim(ReadPbbBase, BaseBerkasKirim):
__tablename__ = "berkas_kirim"
__table_args__ = (PBB_ARGS,)
class ReadSppt(ReadPbbBase, BaseSppt):
__tablename__ = "sppt"
__table_args__ = (PBB_ARGS,)
def get_connection(settings=None):
if not settings:
settings = get_settings()
url = 'pbb_read.url' in settings and settings[
'pbb_read.url'] and 'pbb_read.' or ''
if not url:
raise Exception('"pbb_read.url" belum ada pada settings')
engine = engine_from_config(settings, url)
ReadPbbDBSession.configure(bind=engine)
ReadPbbBase.metadata.bind = engine
# settings['pbb_connected'] = True
def includeme(config):
_logging.info(config)
get_connection()
_logging.info("PBB Mirror Loaded")