Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _add(self, id, scls, data) -> Schema:
name = data['name']
if name in self._name_to_id:
raise errors.SchemaError(
f'{type(scls).__name__} {name!r} is already present '
f'in the schema {self!r}')
data = immu.Map(data)
name_to_id, shortname_to_id, globalname_to_id = self._update_obj_name(
id, scls, None, name)
updates = dict(
id_to_data=self._id_to_data.set(id, data),
id_to_type=self._id_to_type.set(id, scls),
name_to_id=name_to_id,
shortname_to_id=shortname_to_id,
globalname_to_id=globalname_to_id,
refs_to=self._update_refs_to(scls, None, data),
)
if (not isinstance(scls, so.UnqualifiedObject)
and not self.has_module(name.module)):
raise errors.UnknownModuleError(
def __init__(self, db: Database, *, user, query_cache):
self._db = db
self._query_cache_enabled = query_cache
self._user = user
self._config = immutables.Map()
self._modaliases = immutables.Map({None: defines.DEFAULT_MODULE_ALIAS})
# Whenever we are in a transaction that had executed a
# DDL command, we use this cache for compiled queries.
self._eql_to_compiled = lru.LRUMapping(
maxsize=defines._MAX_QUERIES_CACHE)
self._reset_tx_state()
def _unpack_inherited_fields(self, value):
if value is None:
return immu.Map()
else:
return immu.Map(json.loads(value))
@dataclasses.dataclass(frozen=True)
class DDLQuery(BaseQuery):
new_types: FrozenSet[str] = frozenset()
@dataclasses.dataclass(frozen=True)
class TxControlQuery(BaseQuery):
action: TxAction
single_unit: bool
cacheable: bool
modaliases: Optional[immutables.Map]
#############################
@dataclasses.dataclass
class QueryUnit:
dbver: int
sql: Tuple[bytes, ...]
# Status-line for the compiled command; returned to front-end
# in a CommandComplete protocol message if the command is
# executed successfully. When a QueryUnit contains multiple
# EdgeQL queries, the status reflects the last query in the unit.
def __init__(self, db: Database, *, user, query_cache):
self._db = db
self._query_cache_enabled = query_cache
self._user = user
self._config = immutables.Map()
self._modaliases = immutables.Map({None: defines.DEFAULT_MODULE_ALIAS})
# Whenever we are in a transaction that had executed a
# DDL command, we use this cache for compiled queries.
self._eql_to_compiled = lru.LRUMapping(
maxsize=defines._MAX_QUERIES_CACHE)
self._reset_tx_state()
coerce=True,
inheritable=False,
compcoef=0.714,
)
ancestors = SchemaField(
ObjectList,
default=ObjectList,
coerce=True,
inheritable=False,
compcoef=0.999,
)
# Attributes that have been set locally as opposed to inherited.
inherited_fields = SchemaField(
immu.Map,
default=immu.Map(),
inheritable=False,
hashable=False,
)
is_derived = SchemaField(
bool,
default=False, compcoef=0.909)
is_abstract = SchemaField(
bool,
default=False,
inheritable=False, compcoef=0.909)
is_final = SchemaField(
bool,
def _unpack_inherited_fields(self, value):
if value is None:
return immu.Map()
else:
return immu.Map(json.loads(value))
''')
config_spec = config.get_settings()
for script in scripts:
_, sql = compiler.compile_bootstrap_script(
schema, schema, script, single_statement=True)
if debug.flags.bootstrap:
debug.header('Bootstrap')
debug.dump_code(sql, lexer='sql')
config_op_data = await conn.fetchval(sql)
if config_op_data is not None and isinstance(config_op_data, str):
config_op = config.Operation.from_json(config_op_data)
settings = config_op.apply(config_spec, immutables.Map())
config_json = config.to_json(config_spec, settings)
block = dbops.PLTopBlock()
dbops.UpdateMetadata(
dbops.Database(name=edbdef.EDGEDB_TEMPLATE_DB),
{'sysconfig': json.loads(config_json)},
).generate(block)
await _execute_block(conn, block)