Python sqlalchemy.dialects.postgresql 模块，INET 实例源码

def upgrade():
    op.create_table(
        'gateways',
        sa.Column('id', sa.Integer, primary_key=True, autoincrement=True),
        sa.Column('host', INET, nullable=False, unique=True),
        sa.Column('name', sa.String, nullable=True),
        sa.Column('enabled', sa.Boolean, nullable=False, default=True),
        sa.Column('eui', sa.Numeric, nullable=False, unique=True),
        sa.Column('power', sa.Integer, nullable=False),
        sa.Column('port', sa.String, nullable=True),
        sa.Column('latitude', sa.Float, nullable=True),
        sa.Column('longitude', sa.Float, nullable=True),
        sa.Column('created', sa.DateTime(timezone=True), nullable=False),
        sa.Column('updated', sa.DateTime(timezone=True), nullable=False),
        )
    op.add_column('devices',
        sa.Column('devnonce', sa.dialects.postgresql.ARRAY(sa.Integer())))

def upgrade():
    op.create_table(
        'config',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.String, nullable=False),
        sa.Column('listen', INET, nullable=False),
        sa.Column('port', sa.Integer, nullable=False),
        sa.Column('webport', sa.Integer, nullable=False),
        sa.Column('apitoken', sa.String, nullable=False),
        sa.Column('freqband', sa.String, nullable=False),
        sa.Column('netid', sa.Integer, nullable=False),
        sa.Column('duplicateperiod', sa.Integer, nullable=False),
        sa.Column('fcrelaxed', sa.Boolean, nullable=False),
        sa.Column('otaastart', sa.Integer, nullable=False),
        sa.Column('otaaend', sa.Integer, nullable=False),
        sa.Column('macqueueing', sa.Boolean, nullable=False),
        sa.Column('macqueuelimit', sa.Integer, nullable=False),
        sa.Column('adrenable', sa.Boolean, nullable=False),
        sa.Column('adrmargin', sa.Float, nullable=False),
        sa.Column('adrcycletime', sa.Integer, nullable=False),
        sa.Column('adrmessagetime', sa.Integer, nullable=False),
        sa.Column('created', sa.DateTime(timezone=True), nullable=False),
        sa.Column('updated', sa.DateTime(timezone=True), nullable=False),
        )

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    op.create_table('user_log',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('ipaddr', postgresql.INET(), nullable=False),
    sa.Column('timestamp', sa.DateTime(), nullable=False),
    sa.Column('uid', sa.String(length=32), nullable=False),
    sa.Column('method', sa.Enum('GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH', name='http_enum'), nullable=True),
    sa.Column('blueprint', sa.String(length=32), nullable=False),
    sa.Column('path', sa.String(length=128), nullable=False),
    sa.Column('description', sa.String(length=128), nullable=False),
    sa.PrimaryKeyConstraint('id')
    )
    ### end Alembic commands ###

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    op.add_column('node', sa.Column('is_active', sa.Boolean(), server_default='1', nullable=False))
    op.add_column('node', sa.Column('last_ip', postgresql.INET(), nullable=True))
    ### end Alembic commands ###

def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            # Fallback to native PSQL INET
            return dialect.type_descriptor(_INET())
        else:
            return dialect.type_descriptor(self.impl)

def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(postgresql.INET())
        else:
            return dialect.type_descriptor(types.String(39))

def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(postgresql.INET())
        else:
            return dialect.type_descriptor(types.String(43))

def map_column(self, mode: EditMode, request: Request, node: colander.SchemaNode, model: type, name: str, column: Column, column_type: TypeEngine) -> t.Tuple[colander.SchemaType, dict]:
        """Map non-relationship SQLAlchemy column to Colander SchemaNode.

        :return: Tuple(constructed colander.SchemaType, dict of addtional colander.SchemaNode construction arguments)
        """
        logger.debug("Mapping field %s, mode %s, node %s, column %s, column type %s", name, mode, node, column, column_type)

        # Check for autogenerated columns (updated_at)
        if column.onupdate:
            if mode in (EditMode.edit, EditMode.add):
                return TypeOverridesHandling.drop, {}

        # Don't fill default values when added, as they are automatically populated
        if column.default:
            if mode == EditMode.add:
                return TypeOverridesHandling.drop, {}

        # Never add primary keys
        # NOTE: TODO: We need to preserve ids because of nesting mechanism and groupedit widget wants it id
        if column.primary_key:
            # TODO: Looks like column.autoincrement is set True by default, so we cannot use it here
            if mode in (EditMode.edit, EditMode.add):
                return TypeOverridesHandling.drop, {}

        if column.foreign_keys:

            # Handled by relationship mapper
            return TypeOverridesHandling.drop, {}

        elif isinstance(column_type, (PostgreSQLUUID, columns.UUID)):

            # UUID's cannot be22 edited
            if mode in (EditMode.add, EditMode.edit):
                return TypeOverridesHandling.drop, {}

            # But let's show them
            return fields.UUID(), dict(missing=colander.drop, widget=FriendlyUUIDWidget(readonly=True))

        elif isinstance(column_type, Text):
            return colander.String(), dict(widget=deform.widget.TextAreaWidget())
        elif isinstance(column_type, JSONB):
            return JSONValue(), dict(widget=JSONWidget())
        elif isinstance(column_type, (JSONB, columns.JSONB)):
            # Can't edit JSON
            if mode in (EditMode.add, EditMode.edit):
                return TypeOverridesHandling.drop, {}
            return colander.String(), {}
        elif isinstance(column_type, LargeBinary):
            # Can't edit binary
            return TypeOverridesHandling.drop, {}
        elif isinstance(column_type, Geometry):
            # Can't edit geometry
            return TypeOverridesHandling.drop, {}
        elif isinstance(column_type, (INET, columns.INET)):
            return colander.String(), {}
        else:
            # Default mapping / unknown, let the parent handle
            return TypeOverridesHandling.unknown, {}