560 lines
18 KiB
Plaintext
560 lines
18 KiB
Plaintext
/*
|
||
* Copyright (c) 2025 SimcuTeam. All rights reserved.
|
||
* 数据库上下文：连接管理 + 变更提交 + 查询物化。
|
||
*
|
||
* 对齐 EF Core 的用法：
|
||
* class AppDbContext <: DbContext {
|
||
* public init(datasource: Datasource) { super(datasource) }
|
||
* }
|
||
* let db = AppDbContext(ds)
|
||
* db.set<User>().add(user)
|
||
* db.saveChanges()
|
||
*
|
||
* 语义说明（v1，操作队列模式）：add/update/remove 只入队，
|
||
* saveChanges 开启事务按入队顺序逐条执行；执行后清空队列。
|
||
*/
|
||
|
||
package simcu::orm.db
|
||
|
||
import std.collection.*
|
||
import std.database.sql.*
|
||
import std.fs.*
|
||
import std.reflect.*
|
||
import simcu::orm.cli.{CliContext, MigrationCli}
|
||
import simcu::orm.model.*
|
||
import simcu::orm.query.*
|
||
import simcu::orm.sql.*
|
||
import simcu::orm.tracking.*
|
||
import simcu::orm.migrations.{Migrator, Migration}
|
||
|
||
/**
|
||
* 数据库上下文。
|
||
*
|
||
* 基于实例的 CLI 入口（对齐 EF Core）：
|
||
* main(args: Array<String>) {
|
||
* let db = DataContext("pgsql", "Host=..;Database=..;Username=..;Password=..")
|
||
* if (db.cli(args)) { return } // `app orm <命令>` 时执行 CLI 并退出
|
||
* // 正常启动
|
||
* }
|
||
* cli() 的模型来自实例的 DbSet 属性(反射收集)；迁移列表来自子类
|
||
* override migrations()；连接/驱动来自构造参数。
|
||
*/
|
||
public open class DbContext <: CliContext {
|
||
internal let datasource: Datasource
|
||
internal let dialect: SqlDialect
|
||
internal let tracker = ChangeTracker()
|
||
private let _sets = HashMap<String, Any>()
|
||
private var _driverName: String
|
||
private var _connString: String
|
||
|
||
public init(datasource: Datasource, dialect: SqlDialect) {
|
||
this.datasource = datasource
|
||
this.dialect = dialect
|
||
this._driverName = ""
|
||
this._connString = ""
|
||
}
|
||
|
||
public init(datasource: Datasource) {
|
||
this(datasource, OpenGaussDialect())
|
||
}
|
||
|
||
/// 便捷构造：由连接字符串创建 Datasource（驱动构建收敛在 DatasourceFactory）
|
||
public init(connStr: String, dialect: SqlDialect) {
|
||
this(DatasourceFactory.create(connStr), dialect)
|
||
this._driverName = DatasourceFactory.detectDriver(connStr, DatasourceFactory.DEFAULT_DRIVER)
|
||
this._connString = connStr
|
||
}
|
||
|
||
public init(connStr: String) {
|
||
this(connStr, OpenGaussDialect())
|
||
}
|
||
|
||
/// 便捷构造：显式指定驱动类型（如 "pgsql"）+ 连接字符串
|
||
public init(driverName: String, connStr: String, dialect: SqlDialect) {
|
||
this(DatasourceFactory.create(connStr, driverName), dialect)
|
||
this._driverName = DatasourceFactory.normalizeDriverName(driverName)
|
||
this._connString = connStr
|
||
}
|
||
|
||
public init(driverName: String, connStr: String) {
|
||
this(driverName, connStr, OpenGaussDialect())
|
||
}
|
||
|
||
/// 应用提供的迁移列表（子类 override；由 CLI 维护的注册文件提供实例）。
|
||
/// 默认空——add 生成迁移不依赖它，update/list/create 需要。
|
||
public open func migrations(): ArrayList<Migration> {
|
||
ArrayList()
|
||
}
|
||
|
||
/// CLI 入口：`<app> orm <命令>` 时执行迁移 CLI 并返回 true；否则返回 false（正常启动）。
|
||
/// 模型从本实例的 DbSet 属性反射收集，连接/驱动/方言来自构造参数。
|
||
public func cli(args: Array<String>): Bool {
|
||
if (args.size == 0 || args[0] != "orm") {
|
||
return false
|
||
}
|
||
let list = ArrayList<String>()
|
||
for (i in 1..args.size) {
|
||
list.add(args[i])
|
||
}
|
||
MigrationCli.runOn(this, list)
|
||
true
|
||
}
|
||
|
||
/// 反射收集本实例所有 DbSet<T> 的模型（DataContext 的 DbSet 字段/prop 均可）
|
||
public func getModels(): ArrayList<EntityModel> {
|
||
let result = ArrayList<EntityModel>()
|
||
let ti = TypeInfo.of(this)
|
||
// prop（getter 形式）与 var 字段形式都要支持
|
||
for (p in ti.instanceProperties) {
|
||
if (p.typeInfo.name.startsWith("DbSet")) {
|
||
let v = p.getValue(this)
|
||
match (v as DbSetBase) {
|
||
case Some(ds) => result.add(ds.getModel())
|
||
case None => ()
|
||
}
|
||
}
|
||
}
|
||
for (v in ti.instanceVariables) {
|
||
if (v.typeInfo.name.startsWith("DbSet")) {
|
||
let value = v.getValue(this)
|
||
match (value as DbSetBase) {
|
||
case Some(ds) => result.add(ds.getModel())
|
||
case None => ()
|
||
}
|
||
}
|
||
}
|
||
result
|
||
}
|
||
|
||
/// 驱动注册名（构造时归一化；Datasource 构造无连接串时为空）
|
||
public func getDriverName(): String {
|
||
_driverName
|
||
}
|
||
|
||
/// 原始连接字符串（Datasource 构造时为 ""）
|
||
public func getConnString(): String {
|
||
_connString
|
||
}
|
||
|
||
/// CliContext：主数据源
|
||
public func getDatasource(): Datasource {
|
||
datasource
|
||
}
|
||
|
||
/// CliContext：迁移列表（委托子类的 migrations()）
|
||
public func getMigrations(): ArrayList<Migration> {
|
||
migrations()
|
||
}
|
||
|
||
/// CliContext：seed 运行上下文 = 本 DataContext 实例（seed 的 run(ctx) 用它操作 ORM）
|
||
public func getSeedContext(): Any {
|
||
this
|
||
}
|
||
|
||
/// 系统库数据源（数据库存在性检查用）。
|
||
/// 由连接串 + 驱动派生：Database=xxx 换成系统库名（pgsql → postgres，mysql → mysql）。
|
||
/// 无连接串或驱动不支持（sqlite 无系统库）时返回 None。
|
||
public func getAdminDatasource(): ?Datasource {
|
||
if (_connString.isEmpty() || _driverName.isEmpty()) {
|
||
return None
|
||
}
|
||
if (_driverName == "sqlite") {
|
||
return None
|
||
}
|
||
let adminDb = if (_driverName == "mysql") { "mysql" } else { "postgres" }
|
||
var result = ""
|
||
for (seg in _connString.split(";")) {
|
||
let t = seg.trimAscii()
|
||
if (t.isEmpty()) {
|
||
continue
|
||
}
|
||
if (t.toAsciiLower().startsWith("database=")) {
|
||
result += "Database=${adminDb};"
|
||
} else {
|
||
result += t
|
||
result += ";"
|
||
}
|
||
}
|
||
Some(DatasourceFactory.create(result, _driverName))
|
||
}
|
||
|
||
/// CliContext：应用所有迁移（委托 migrate）
|
||
public func migrateAll(migrations: ArrayList<Migration>): Int64 {
|
||
migrate(migrations)
|
||
}
|
||
|
||
/// 获取实体的 DbSet（同一实体类型复用同一个 DbSet 实例）
|
||
public func set<T>(): DbSet<T> {
|
||
let key = TypeInfo.of<T>().toString()
|
||
if (let Some(existing) <- _sets.get(key)) {
|
||
return (existing as DbSet<T>).getOrThrow()
|
||
}
|
||
let ds = DbSet<T>(this, ModelCache.get<T>())
|
||
_sets[key] = ds
|
||
ds
|
||
}
|
||
|
||
/// 待提交的变更条数
|
||
public func pendingCount(): Int64 {
|
||
tracker.count()
|
||
}
|
||
|
||
/// 当前方言（子类/应用可读取）
|
||
public func getDialect(): SqlDialect {
|
||
dialect
|
||
}
|
||
|
||
/// 应用迁移（对齐 EF Core Database.Migrate），返回本次应用的数量
|
||
public func migrate(migrations: ArrayList<Migration>): Int64 {
|
||
Migrator(datasource, dialect).migrate(migrations)
|
||
}
|
||
|
||
/// 无参便捷版：应用子类 migrations() 提供的全部迁移（`db.migrate()` 即运行 update）
|
||
public func migrate(): Int64 {
|
||
migrate(migrations())
|
||
}
|
||
|
||
/// 确认当前连接串指向的数据库是否存在（连系统库查询，不建库）。
|
||
/// 需带连接串构造（_connString 非空）；sqlite 文件库返回文件是否存在。
|
||
public func databaseExists(): Bool {
|
||
if (_connString.isEmpty()) {
|
||
return false
|
||
}
|
||
if (_driverName == "sqlite") {
|
||
return exists(Path(sqliteFilePath()))
|
||
}
|
||
let dbName = databaseNameFromConnString()
|
||
if (dbName.isEmpty()) {
|
||
return false
|
||
}
|
||
match (getAdminDatasource()) {
|
||
case None => false
|
||
case Some(admin) =>
|
||
let conn = admin.connect()
|
||
try {
|
||
let stmt = conn.prepareStatement(dialect.databaseExistsSql())
|
||
try {
|
||
stmt.set<Any>(0, dbName)
|
||
let rs = stmt.query()
|
||
rs.next()
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
} finally {
|
||
conn.close()
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 是否存在未应用的迁移（库未创建或历史表不存在视为有待应用迁移）
|
||
public func hasPendingMigrations(): Bool {
|
||
let list = migrations()
|
||
if (list.size == 0) {
|
||
return false
|
||
}
|
||
try {
|
||
let migrator = Migrator(datasource, dialect)
|
||
let applied = migrator.appliedMigrationIds()
|
||
for (m in list) {
|
||
if (!applied.contains(m.migrationId)) {
|
||
return true
|
||
}
|
||
}
|
||
false
|
||
} catch (_: Exception) {
|
||
// 数据库/历史表不存在 → 有待应用迁移
|
||
true
|
||
}
|
||
}
|
||
|
||
/// 提交全部待处理变更（事务内按入队顺序执行），返回影响的实体数
|
||
public func saveChanges(): Int64 {
|
||
var count: Int64 = 0
|
||
if (tracker.count() == 0) {
|
||
return 0
|
||
}
|
||
let conn = datasource.connect()
|
||
let tx = conn.createTransaction()
|
||
tx.begin()
|
||
try {
|
||
for (entry in tracker.getEntries()) {
|
||
match (entry.state) {
|
||
case EntityState.Added => count += insertOne(conn, entry)
|
||
case EntityState.Modified => count += updateOne(conn, entry)
|
||
case EntityState.Deleted => count += deleteOne(conn, entry)
|
||
case _ => ()
|
||
}
|
||
}
|
||
tx.commit()
|
||
tracker.clear()
|
||
count
|
||
} catch (e: Exception) {
|
||
try {
|
||
tx.rollback()
|
||
} catch (_) {
|
||
()
|
||
}
|
||
throw e
|
||
} finally {
|
||
conn.close()
|
||
}
|
||
}
|
||
|
||
internal func getTracker(): ChangeTracker {
|
||
tracker
|
||
}
|
||
|
||
/// 执行查询并物化实体（query 层回调）
|
||
internal func query<T>(model: EntityModel, sql: String, params: ArrayList<Any>): ArrayList<T> {
|
||
let conn = datasource.connect()
|
||
try {
|
||
let stmt = conn.prepareStatement(sql)
|
||
try {
|
||
ParamBinder.bind(stmt, params)
|
||
let rs = stmt.query()
|
||
let list = ArrayList<T>()
|
||
while (rs.next()) {
|
||
let instance = materialize(rs, model)
|
||
list.add((instance as T).getOrThrow())
|
||
}
|
||
list
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
} finally {
|
||
conn.close()
|
||
}
|
||
}
|
||
|
||
/// 执行 COUNT 查询（query 层回调）
|
||
internal func count(sql: String, params: ArrayList<Any>): Int64 {
|
||
let conn = datasource.connect()
|
||
try {
|
||
let stmt = conn.prepareStatement(sql)
|
||
try {
|
||
ParamBinder.bind(stmt, params)
|
||
let rs = stmt.query()
|
||
if (rs.next()) {
|
||
return rs.getOrNull<Int64>(0).getOrThrow()
|
||
}
|
||
0
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
} finally {
|
||
conn.close()
|
||
}
|
||
}
|
||
|
||
/// 结果集 → 实体实例（列顺序与 model.properties 一致）
|
||
internal func materialize(rs: QueryResult, model: EntityModel): Any {
|
||
let instance = model.createInstance()
|
||
var i = 0
|
||
for (p in model.properties) {
|
||
let v = ValueReader.read(rs, i, p.fullTypeName())
|
||
i += 1
|
||
if (let Some(val) <- v) {
|
||
model.setValue(instance, p, val)
|
||
}
|
||
}
|
||
instance
|
||
}
|
||
|
||
private func insertOne(conn: Connection, entry: EntityEntry): Int64 {
|
||
let model = modelOf(entry.entity)
|
||
let insertCols = ArrayList<String>()
|
||
let values = ArrayList<Any>()
|
||
for (p in model.properties) {
|
||
if (p.autoIncrement) {
|
||
continue
|
||
}
|
||
var v = model.getValue(entry.entity, p)
|
||
// 客户端生成主键（String）：为空则生成唯一串
|
||
if (p.clientGenerated && isBlankString(v)) {
|
||
v = GuidUtil.generate()
|
||
model.setValue(entry.entity, p, v)
|
||
}
|
||
insertCols.add(p.columnName)
|
||
values.add(v)
|
||
}
|
||
let kp = model.keyProperty.getOrThrow()
|
||
let returning = kp.autoIncrement
|
||
let sql = dialect.buildInsert(model.tableName, insertCols, kp.columnName, returning)
|
||
let stmt = conn.prepareStatement(sql)
|
||
try {
|
||
ParamBinder.bind(stmt, values)
|
||
if (returning) {
|
||
let rs = stmt.query()
|
||
if (rs.next()) {
|
||
let v = ValueReader.read(rs, 0, kp.fullTypeName())
|
||
if (let Some(kv) <- v) {
|
||
model.setKeyValue(entry.entity, kv)
|
||
}
|
||
}
|
||
} else {
|
||
stmt.update()
|
||
}
|
||
1
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
}
|
||
|
||
private func updateOne(conn: Connection, entry: EntityEntry): Int64 {
|
||
let model = modelOf(entry.entity)
|
||
let setCols = ArrayList<String>()
|
||
let values = ArrayList<Any>()
|
||
for (p in model.properties) {
|
||
if (p.isKey) {
|
||
continue
|
||
}
|
||
setCols.add(p.columnName)
|
||
values.add(model.getValue(entry.entity, p))
|
||
}
|
||
let kp = model.keyProperty.getOrThrow()
|
||
values.add(model.getKeyValue(entry.entity))
|
||
let sql = dialect.buildUpdate(model.tableName, setCols, kp.columnName)
|
||
let stmt = conn.prepareStatement(sql)
|
||
try {
|
||
ParamBinder.bind(stmt, values)
|
||
stmt.update()
|
||
1
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
}
|
||
|
||
private func deleteOne(conn: Connection, entry: EntityEntry): Int64 {
|
||
let model = modelOf(entry.entity)
|
||
let kp = model.keyProperty.getOrThrow()
|
||
let sql = dialect.buildDelete(model.tableName, kp.columnName)
|
||
let stmt = conn.prepareStatement(sql)
|
||
try {
|
||
stmt.set<Any>(0, model.getKeyValue(entry.entity))
|
||
stmt.update()
|
||
1
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
}
|
||
|
||
private func modelOf(entity: Any): EntityModel {
|
||
ModelCache.getFor(TypeInfo.of(entity))
|
||
}
|
||
|
||
private static func isBlankString(v: Any): Bool {
|
||
if (let s: String <- v) {
|
||
return s.isEmpty()
|
||
}
|
||
false
|
||
}
|
||
|
||
/// 从连接串解析数据库名：
|
||
/// URL 格式（postgres://user:pass@host:port/dbname?params）→ 路径段（去 query）；
|
||
/// ADO 风格（Host=..;Database=xxx;..）→ Database 值。
|
||
/// 解析不到返回 ""。
|
||
private func databaseNameFromConnString(): String {
|
||
if (_connString.contains("://")) {
|
||
let afterScheme = _connString.indexOf("://").getOrThrow() + 3
|
||
let rest = _connString[afterScheme..]
|
||
if (let Some(slash) <- rest.indexOf("/")) {
|
||
var path = rest[slash + 1..]
|
||
if (let Some(q) <- path.indexOf("?")) {
|
||
path = path[..q]
|
||
}
|
||
return path
|
||
}
|
||
return ""
|
||
}
|
||
for (seg in _connString.split(";")) {
|
||
let t = seg.trimAscii()
|
||
if (t.toAsciiLower().startsWith("database=")) {
|
||
return t[t.indexOf("=").getOrThrow() + 1..]
|
||
}
|
||
}
|
||
""
|
||
}
|
||
|
||
/// sqlite 连接串 → 数据库文件路径（去 sqlite:// 前缀与 query 参数）
|
||
private func sqliteFilePath(): String {
|
||
var p = _connString
|
||
if (p.startsWith("sqlite://")) {
|
||
p = p[9..]
|
||
} else if (p.startsWith("sqlite:")) {
|
||
p = p[7..]
|
||
}
|
||
if (let Some(q) <- p.indexOf("?")) {
|
||
p = p[..q]
|
||
}
|
||
p
|
||
}
|
||
}
|
||
|
||
/**
|
||
* DbSet 的非泛型访问基类：供 ORM CLI 通过反射拿到模型（EntityModel）。
|
||
*/
|
||
public abstract class DbSetBase {
|
||
public func getModel(): EntityModel
|
||
}
|
||
|
||
/**
|
||
* 实体的数据集（对齐 EF Core DbSet）。
|
||
*/
|
||
public class DbSet<T> <: DbSetBase {
|
||
private let _context: DbContext
|
||
private let _model: EntityModel
|
||
private let _tracker: ChangeTracker
|
||
|
||
public init(context: DbContext, model: EntityModel) {
|
||
this._context = context
|
||
this._model = model
|
||
this._tracker = context.getTracker()
|
||
}
|
||
|
||
public override func getModel(): EntityModel {
|
||
_model
|
||
}
|
||
|
||
/// 入队新增
|
||
public func add(entity: T): Unit {
|
||
_tracker.add(entity, EntityState.Added)
|
||
}
|
||
|
||
/// 入队修改（UPDATE 全部非主键列）
|
||
public func update(entity: T): Unit {
|
||
_tracker.add(entity, EntityState.Modified)
|
||
}
|
||
|
||
/// 入队删除
|
||
public func remove(entity: T): Unit {
|
||
_tracker.add(entity, EntityState.Deleted)
|
||
}
|
||
|
||
/// 构建查询
|
||
public func query(): QueryBuilder<T> {
|
||
QueryBuilder<T>(_model, _context.getDialect(),
|
||
{ sql, params => _context.query<T>(_model, sql, params) },
|
||
{ sql, params => _context.count(sql, params) })
|
||
}
|
||
|
||
/// 按主键查询（无结果返回 None）
|
||
public func find(key: Any): ?T {
|
||
let kp = _model.keyProperty.getOrThrow()
|
||
let list = query().filter(kp.columnName, "=", key).take(1).toList()
|
||
if (list.size > 0) {
|
||
return Some(list[0])
|
||
}
|
||
None
|
||
}
|
||
|
||
/// 全表查询
|
||
public func toList(): ArrayList<T> {
|
||
query().toList()
|
||
}
|
||
|
||
/// 全表计数
|
||
public func count(): Int64 {
|
||
query().count()
|
||
}
|
||
}
|